helper-secure-views-to-dynamic-tables/secured_views_to_dynamic_tables.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "35d819f8-c137-4e99-9efd-ebe1d56458af",
   "metadata": {
    "collapsed": false,
    "name": "Overview"
   },
   "source": [
    "# Replace Secured Views in Shares with Dynamic Tables\n",
    "\n",
    "## Overview\n",
    "\n",
    "This notebook is designed to streamline the management and utilization of secured views within Snowflake shares by converting them into dynamic tables. The process helps in enhancing data freshness, enabling more efficient data management practices, and optimizing query performance.\n",
    "\n",
    "## Process\n",
    "\n",
    "1. **Identify Secured Views in Database**: \n",
    "\n",
    "    The first step involves scanning the Snowflake database to identify all secured views. Secured views are those that have restricted access, ensuring that sensitive data is protected and only accessible to authorized users.\n",
    "\n",
    "    We query the `SNOWFLAKE.ACCOUNT_USAGE.VIEWS` system view to retrieve a list of all views in the account, filtering specifically for those marked as secured. This helps in segregating the views that are crucial for sensitive data handling.\n",
    "  \n",
    "2. **Extract Object Dependencies**:\n",
    "\n",
    "    Once the secured views are identified, the next step is to extract their dependencies. This includes understanding which tables, functions, or other database objects the views depend on.\n",
    "\n",
    "    This involves querying the `SNOWFLAKE.ACCOUNT_USAGE.OBJECT_DEPENDENCIES` system function to map out all underlying objects that the secured views interact with. This mapping is essential for ensuring that all dependent objects are considered when converting the views into dynamic tables.\n",
    "\n",
    "3. **Qualify Secured Views for conversion**: \n",
    "\n",
    "    Not all secured views may be suitable for conversion to dynamic tables. This step evaluates each secured view to determine if it meets the criteria for conversion based on factors such as complexity, dependencies, and potential performance benefits.\n",
    "\n",
    "    Apply a set of predefined rules or checks to assess which views are ideal candidates for conversion. This might include checking for views that do not involve overly complex SQL operations or those that are frequently accessed and would benefit most from dynamic table features.\n",
    "\n",
    "4. **Create conversion Code**: \n",
    "\n",
    "    For views that qualify, generate the SQL code needed to convert these secured views into dynamic tables. Dynamic tables in Snowflake allow for automatic data management features such as clustering and micro-partitioning, which improve query performance and data retrieval speed.\n",
    "\n",
    "![process_flow](https://lh3.googleusercontent.com/d/1hQMggZQ706k9FSApWXMdFMCm4TIx1-yl)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b552f21-11dd-403c-9468-417862ef342a",
   "metadata": {
    "collapsed": false,
    "name": "Session_Label"
   },
   "source": [
    "# Initialize Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3775908f-ca36-4846-8f38-5adca39217f2",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "Initialize_Session"
   },
   "outputs": [],
   "source": [
    "# Import python packages\n",
    "import streamlit as st\n",
    "import pandas as pd\n",
    "import re\n",
    "\n",
    "# We can also use Snowpark for our analyses!\n",
    "from snowflake.snowpark.context import get_active_session\n",
    "session = get_active_session()\n",
    "\n",
    "#tag session\n",
    "session.sql(f\"\"\"ALTER SESSION SET QUERY_TAG = '{{\"origin\":\"sf_sit\",\"name\":\"dt_secure_view_conversion_task\",\"version\":{{\"major\":1, \"minor\":0}},\"attributes\":\"session_tag\"}}'\"\"\").collect()\n",
    "\n",
    "st.success(f\"Session initialized 🎉\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "35ca3e4d-a39f-4a37-bbe2-e383ce76de33",
   "metadata": {
    "collapsed": false,
    "name": "Utils_Label"
   },
   "source": [
    "# Functions used in Notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7e00ad25-024b-4b68-86ee-a06c8a64f807",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "Function_Definitions"
   },
   "outputs": [],
   "source": [
    "\n",
    "def get_shares(session):\n",
    "    \"\"\"\n",
    "    Retrieves all shares from Snowflake.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :return: Pandas DataFrame of shares\n",
    "    \"\"\"\n",
    "    shares_query = \"SHOW SHARES;\"\n",
    "    shares_rows = session.sql(shares_query).collect()\n",
    "    df_shares = pd.DataFrame(shares_rows)\n",
    "    return df_shares\n",
    "\n",
    "\n",
    "def filter_shares_by_database(df_shares, database_name):\n",
    "    \"\"\"\n",
    "    Filters the shares to only include those related to the given database.\n",
    "    \n",
    "    :param df_shares: DataFrame containing all shares\n",
    "    :param database_name: Name of the database to filter on\n",
    "    :return: Filtered DataFrame of shares related to the database\n",
    "    \"\"\"\n",
    "    return df_shares[df_shares['database_name'] == database_name.upper()]\n",
    "\n",
    "\n",
    "def get_secure_views(session, database_name):\n",
    "    \"\"\"\n",
    "    Retrieves all secure views from the given database along with their view definitions.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param database_name: Name of the database\n",
    "    :return: Pandas DataFrame of secure views with their definitions\n",
    "    \"\"\"\n",
    "    secure_views_query = f\"\"\"\n",
    "        SELECT TABLE_SCHEMA, TABLE_NAME, VIEW_DEFINITION\n",
    "        FROM SNOWFLAKE.ACCOUNT_USAGE.VIEWS\n",
    "        WHERE TABLE_CATALOG = '{database_name}'\n",
    "        AND DELETED IS NULL\n",
    "        and IS_SECURE = 'YES';\n",
    "    \"\"\"\n",
    "    return session.sql(secure_views_query).to_pandas()\n",
    "\n",
    "\n",
    "def get_shared_views_for_share(session, share_name):\n",
    "    \"\"\"\n",
    "    Retrieves all shared views for a given share.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param share_name: Name of the share\n",
    "    :return: List of tuples containing schema and view names, or an empty list if no views are found\n",
    "    \"\"\"\n",
    "    desc_share_query = f\"DESC SHARE {share_name};\"\n",
    "    share_details_rows = session.sql(desc_share_query).collect()\n",
    "    \n",
    "    # If no rows are returned, return an empty list\n",
    "    if not share_details_rows:\n",
    "        return []\n",
    "\n",
    "    df_share_details = pd.DataFrame(share_details_rows)\n",
    "    \n",
    "    # Filter for views and split the 'name' column into database, schema, and view\n",
    "    df_share_views = df_share_details[df_share_details['kind'] == 'VIEW'].copy()\n",
    "\n",
    "    # If no views are found, return an empty list\n",
    "    if df_share_views.empty:\n",
    "        return []\n",
    "    \n",
    "    df_share_views[['DATABASE', 'SCHEMA', 'VIEW']] = df_share_views['name'].str.split('.', expand=True)\n",
    "    \n",
    "    return df_share_views[['SCHEMA', 'VIEW']].values.tolist()\n",
    "\n",
    "\n",
    "\n",
    "def get_all_shared_views(session, df_shares_filtered):\n",
    "    \"\"\"\n",
    "    Loops through all shares and retrieves shared views, including the share name.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param df_shares_filtered: Filtered DataFrame of shares related to the database\n",
    "    :return: List of tuples containing share name, schema name, and view name for all shares\n",
    "    \"\"\"\n",
    "    shared_views = []\n",
    "    for share in df_shares_filtered['name']:\n",
    "        # Get shared views for the current share\n",
    "        views_for_share = get_shared_views_for_share(session, share)\n",
    "        \n",
    "        # Add the share name to each entry in the result\n",
    "        for schema_name, view_name in views_for_share:\n",
    "            shared_views.append((share, schema_name, view_name))\n",
    "    \n",
    "    return shared_views\n",
    "\n",
    "\n",
    "\n",
    "def check_if_views_are_shared(df_secure_views, shared_views):\n",
    "    \"\"\"\n",
    "    Adds columns to df_secure_views indicating if the view is part of any shared views and the corresponding share name.\n",
    "    \n",
    "    :param df_secure_views: DataFrame of secure views\n",
    "    :param shared_views: List of tuples containing share name, schema name, and view name\n",
    "    :return: DataFrame with 'IS_SHARED' and 'SHARE_NAME' columns added\n",
    "    \"\"\"\n",
    "    # Convert shared_views to a set of tuples for efficient lookups\n",
    "    shared_views_dict = { (schema_name, view_name): share_name for share_name, schema_name, view_name in shared_views }\n",
    "    \n",
    "    def get_share_info(row):\n",
    "        # Check if the (schema, view) exists in the shared_views dictionary and return the share name\n",
    "        schema_view_tuple = (row['TABLE_SCHEMA'], row['TABLE_NAME'])\n",
    "        if schema_view_tuple in shared_views_dict:\n",
    "            return True, shared_views_dict[schema_view_tuple]\n",
    "        return False, None\n",
    "\n",
    "    # Apply the function to add 'IS_SHARED' and 'SHARE_NAME' columns\n",
    "    df_secure_views[['IS_SHARED', 'SHARE_NAME']] = df_secure_views.apply(\n",
    "        lambda row: pd.Series(get_share_info(row)),\n",
    "        axis=1\n",
    "    )\n",
    "\n",
    "    return df_secure_views\n",
    "\n",
    "\n",
    "def get_ddl_for_shared_views(session, df_secure_views):\n",
    "    \"\"\"\n",
    "    Loops through each secure view that is being shared and retrieves the DDL for it.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param df_secure_views: DataFrame containing secure views with the 'IS_SHARED' column\n",
    "    :return: Dictionary where keys are view names and values are their DDL statements\n",
    "    \"\"\"\n",
    "    ddl_statements = {}\n",
    "\n",
    "    # Filter only the secure views that are being shared\n",
    "    shared_views = df_secure_views[df_secure_views['IS_SHARED'] == True]\n",
    "\n",
    "    # Loop through each shared view and get its DDL\n",
    "    for index, row in shared_views.iterrows():\n",
    "        schema = row['TABLE_SCHEMA']\n",
    "        view_name = row['TABLE_NAME']\n",
    "        \n",
    "        # Use GET_DDL function to get the DDL for the view\n",
    "        get_ddl_query = f\"SELECT GET_DDL('VIEW', '{schema}.{view_name}');\"\n",
    "        \n",
    "        # Execute the query and fetch the DDL\n",
    "        ddl_result = session.sql(get_ddl_query).collect()\n",
    "        ddl_statements[f\"{schema}.{view_name}\"] = ddl_result[0][0]\n",
    "    \n",
    "    return ddl_statements    \n",
    "\n",
    "def check_object_type(session, database_name, schema_name, object_name):\n",
    "    \"\"\"\n",
    "    Checks whether a given object is a table or a view in Snowflake.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param database_name: Name of the database\n",
    "    :param schema_name: Name of the schema\n",
    "    :param object_name: Name of the object (table/view)\n",
    "    :return: String indicating whether the object is a 'TABLE', 'VIEW', or 'UNKNOWN'\n",
    "    \"\"\"\n",
    "    # Check if the object is a table\n",
    "    table_query = f\"\"\"\n",
    "        SELECT COUNT(*) FROM {database_name}.INFORMATION_SCHEMA.TABLES\n",
    "        WHERE TABLE_SCHEMA = '{schema_name}' AND TABLE_NAME = '{object_name}';\n",
    "    \"\"\"\n",
    "    table_result = session.sql(table_query).collect()\n",
    "    \n",
    "    if table_result[0][0] > 0:\n",
    "        return 'TABLE'\n",
    "    \n",
    "    # Check if the object is a view\n",
    "    view_query = f\"\"\"\n",
    "        SELECT COUNT(*) FROM {database_name}.INFORMATION_SCHEMA.VIEWS\n",
    "        WHERE TABLE_SCHEMA = '{schema_name}' AND TABLE_NAME = '{object_name}';\n",
    "    \"\"\"\n",
    "    view_result = session.sql(view_query).collect()\n",
    "    \n",
    "    if view_result[0][0] > 0:\n",
    "        return 'VIEW'\n",
    "    \n",
    "    # If neither, return 'UNKNOWN'\n",
    "    return 'UNKNOWN'\n",
    "\n",
    "\n",
    "def analyze_dependencies_with_cluster_keys(session, ddl_dict):\n",
    "    \"\"\"\n",
    "    Analyzes the dependencies of each view in the DDL dictionary to determine whether the dependencies are tables or views,\n",
    "    and if a table, also checks for clustering keys.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param ddl_dict: Dictionary where keys are view names and values are their DDL statements\n",
    "    :return: Dictionary where keys are view names and values are lists of tuples (dependency, type, cluster_keys)\n",
    "    \"\"\"\n",
    "    view_analysis = {}\n",
    "    \n",
    "    # Loop through each view in the dictionary\n",
    "    for view_name, ddl_statement in ddl_dict.items():\n",
    "        # Extract the dependencies\n",
    "        dependencies = get_object_dependencies(ddl_statement)\n",
    "        \n",
    "        # Initialize a list to hold the analyzed dependencies for this view\n",
    "        analyzed_dependencies = []\n",
    "        \n",
    "        # Analyze each dependency\n",
    "        for dep in dependencies:\n",
    "            # Split the fully qualified name (e.g., SCHEMA.OBJECT)\n",
    "            parts = dep.split('.')\n",
    "            \n",
    "            if len(parts) == 3:  # Ensure the object is fully qualified (database.schema.object)\n",
    "                database_name, schema_name, object_name = parts\n",
    "                # Check if it's a table or a view\n",
    "                object_type = check_object_type(session, database_name, schema_name, object_name)\n",
    "                \n",
    "                # Initialize an empty list for cluster keys\n",
    "                cluster_keys = []\n",
    "                \n",
    "                # If it's a table, retrieve the cluster keys\n",
    "                if object_type == 'TABLE':\n",
    "                    cluster_keys = get_cluster_keys(session, database_name, schema_name, object_name)\n",
    "                \n",
    "                # Append the dependency along with its type and cluster keys\n",
    "                analyzed_dependencies.append((dep, object_type, cluster_keys))\n",
    "        \n",
    "        # Store the result in the view_analysis dictionary\n",
    "        view_analysis[view_name] = analyzed_dependencies\n",
    "    \n",
    "    return view_analysis\n",
    "\n",
    "def get_cluster_keys(session, database_name, schema_name, table_name):\n",
    "    \"\"\"\n",
    "    Retrieves the cluster keys for a given table in Snowflake by using the SHOW TABLES LIKE command.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param database_name: Name of the database\n",
    "    :param schema_name: Name of the schema\n",
    "    :param table_name: Name of the table\n",
    "    :return: List of cluster keys, if present\n",
    "    \"\"\"\n",
    "    # Query to show information about the table (including clustering keys)\n",
    "    show_table_query = f\"\"\"\n",
    "        SHOW TABLES LIKE '{table_name}' IN SCHEMA {database_name}.{schema_name};\n",
    "    \"\"\"\n",
    "    \n",
    "    # Execute the query and collect the results\n",
    "    table_info_result = session.sql(show_table_query).collect()\n",
    "\n",
    "    # Check if the table has clustering keys in the 'cluster_by' column\n",
    "    if table_info_result:\n",
    "        cluster_by_value = table_info_result[0]['cluster_by']\n",
    "        if cluster_by_value:\n",
    "            # Return the clustering keys as a list (split by commas if there are multiple keys)\n",
    "            return [key.strip() for key in cluster_by_value.replace('LINEAR(', '').replace(')', '').split(',')]\n",
    "    \n",
    "    return []  # Return an empty list if no clustering keys are found\n",
    "    \n",
    "def get_dynamic_table_ddl_for_views(view_definitions, dependencies_dict, database_name, target_lag, refresh_mode):\n",
    "    \"\"\"\n",
    "    Constructs the DDL for dynamic tables based on shared secure views, their dependencies, and clustering keys.\n",
    "    Drops the original views before creating the dynamic table and adds the dynamic table to the share.\n",
    "    \n",
    "    :param view_definitions: DataFrame containing view names, view definitions, and share information.\n",
    "    :param dependencies_dict: Dictionary where keys are view names and values are lists of dependency details.\n",
    "    :param database_name: The database name where dynamic tables will be created.\n",
    "    :param target_lag: The target lag for the dynamic table.\n",
    "    :param refresh_mode: The refresh mode for the dynamic table.\n",
    "    :return: Dictionary where keys are view names and values are the generated DDL statements for dynamic tables.\n",
    "    \"\"\"\n",
    "    dynamic_table_ddls = {}\n",
    "\n",
    "    for index, row in view_definitions.iterrows():\n",
    "        view_name = row['TABLE_NAME']\n",
    "        view_ddl = row['VIEW_DEFINITION']\n",
    "        share_name = row['SHARE_NAME']  # Assuming this column contains the share name\n",
    "\n",
    "        # Extract only the SELECT statement from the view DDL\n",
    "        select_start_idx = view_ddl.lower().find(\"select\")\n",
    "        if select_start_idx == -1:\n",
    "            print(f\"Skipping view {view_name}: No SELECT statement found in DDL.\")\n",
    "            continue\n",
    "        select_statement = view_ddl[select_start_idx:].strip()\n",
    "\n",
    "        # Initialize clustering clause\n",
    "        cluster_by_clause = \"\"\n",
    "\n",
    "        # Check if there are dependencies for this view and create CLUSTER BY clause\n",
    "        if view_name in dependencies_dict:\n",
    "            cluster_keys = [dep['REFERENCED_OBJECT_NAME'] for dep in dependencies_dict[view_name] if dep['REFERENCED_OBJECT_DOMAIN'] == 'TABLE']\n",
    "            if cluster_keys:\n",
    "                cluster_by_clause = f\" CLUSTER BY ({', '.join(cluster_keys)})\"\n",
    "\n",
    "        # Generate the SQL for dropping the view if it exists\n",
    "        drop_view_sql = f\"DROP VIEW IF EXISTS {database_name}.{view_name};\"\n",
    "\n",
    "        # Generate the SQL for creating the dynamic table using the SELECT statement and cluster keys\n",
    "        create_table_sql = f\"\"\"\n",
    "        CREATE OR REPLACE DYNAMIC TABLE {database_name}.{view_name}\n",
    "        COMMENT = '{{\"origin\":\"sf_sit\",\"name\":\"dt_secure_view_conversion_task,\"version\":{{\"major\":1, \"minor\":1}},\"attributes\":{{\"table\":\"{view_name}\", \"type\":\"dynamictable\"}}}}'\n",
    "        REFRESH_MODE = {refresh_mode}\n",
    "        TARGET_LAG = '{target_lag}'\n",
    "        WAREHOUSE = 'COHORT_BUILDER_LOAD_WH'\n",
    "        {cluster_by_clause}\n",
    "        AS\n",
    "        {select_statement}\n",
    "        \"\"\"\n",
    "\n",
    "        # Generate the SQL for adding the dynamic table to the share\n",
    "        add_to_share_sql = f\"ALTER SHARE {share_name} ADD TABLE {database_name}.{view_name};\"\n",
    "\n",
    "        # Combine all the SQL into a single script for this view\n",
    "        full_ddl_script = f\"\"\"\n",
    "        {drop_view_sql}\n",
    "        \n",
    "        {create_table_sql.strip()}\n",
    "        \n",
    "        {add_to_share_sql}\n",
    "        \"\"\"\n",
    "\n",
    "        # Add the generated DDL to the dictionary\n",
    "        dynamic_table_ddls[view_name] = full_ddl_script.strip()\n",
    "\n",
    "    return dynamic_table_ddls\n",
    "\n",
    "\n",
    "def remove_comments(sql_code):\n",
    "    \"\"\"\n",
    "    Removes SQL comments (both single-line and multi-line) from the given SQL code.\n",
    "    \n",
    "    :param sql_code: The SQL code as a string.\n",
    "    :return: The SQL code with comments removed.\n",
    "    \"\"\"\n",
    "    # Remove single-line comments (starting with --)\n",
    "    sql_code = re.sub(r'--.*', '', sql_code)\n",
    "    \n",
    "    # Remove multi-line comments (starting with /* and ending with */)\n",
    "    sql_code = re.sub(r'/\\*.*?\\*/', '', sql_code, flags=re.DOTALL)\n",
    "    \n",
    "    return sql_code\n",
    "\n",
    "\n",
    "def extract_query_from_view_ddl(view_ddl):\n",
    "    \"\"\"\n",
    "    Extracts the query (including CTEs if present) from a view's DDL.\n",
    "    \n",
    "    :param view_ddl: The DDL statement of the view as a string\n",
    "    :return: The query as a string (can start with SELECT or WITH) or None if no valid query is found\n",
    "    \"\"\"\n",
    "    # Ensure the DDL is in lowercase to make it case-insensitive\n",
    "    ddl_lower = view_ddl.lower()\n",
    "\n",
    "    # Find the starting index of the SELECT or WITH statement\n",
    "    select_start_idx = ddl_lower.find(\"select\")\n",
    "    with_start_idx = ddl_lower.find(\"with\")\n",
    "\n",
    "    # Determine which comes first, SELECT or WITH, if either exists\n",
    "    if select_start_idx == -1 and with_start_idx == -1:\n",
    "        print(\"No SELECT or WITH statement found in the view DDL.\")\n",
    "        return None\n",
    "\n",
    "    # Use the position of the first occurrence of either SELECT or WITH\n",
    "    query_start_idx = min(\n",
    "        idx for idx in [select_start_idx, with_start_idx] if idx != -1\n",
    "    )\n",
    "\n",
    "    # Extract everything starting from the SELECT or WITH statement\n",
    "    query_statement = view_ddl[query_start_idx:].strip()\n",
    "\n",
    "    # Optionally, remove any trailing semicolons if they exist\n",
    "    if query_statement.endswith(\";\"):\n",
    "        query_statement = query_statement[:-1]\n",
    "\n",
    "    return query_statement\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "def create_dependency_dict(session, shared_views, database_name):\n",
    "    \"\"\"\n",
    "    Creates a clean dependency dictionary for each shared view, containing \n",
    "    REFERENCED_OBJECT_DOMAIN, REFERENCED_OBJECT_NAME, along with schema and \n",
    "    catalog information.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param shared_views: Pandas DataFrame containing shared views\n",
    "    :return: Dictionary with view names as keys, and references with schema and \n",
    "             catalog information\n",
    "    \"\"\"\n",
    "    dependency_dict = {}\n",
    "\n",
    "    # Loop through each row in the shared_views DataFrame\n",
    "    for index, row in shared_views.iterrows():\n",
    "        view_name = row['TABLE_NAME']\n",
    "        schema_name = row['TABLE_SCHEMA']\n",
    "        database_name = database_name\n",
    "\n",
    "        # Query to find dependencies for each shared view, selecting the required columns\n",
    "        dependency_query = f\"\"\"\n",
    "        SELECT REFERENCED_OBJECT_DOMAIN, REFERENCED_OBJECT_NAME, REFERENCED_SCHEMA, REFERENCED_DATABASE\n",
    "        FROM SNOWFLAKE.ACCOUNT_USAGE.OBJECT_DEPENDENCIES\n",
    "        WHERE REFERENCING_OBJECT_NAME = '{view_name}'\n",
    "        AND REFERENCING_SCHEMA = '{schema_name}'\n",
    "        AND REFERENCING_DATABASE = '{database_name}';\n",
    "        \"\"\"\n",
    "        \n",
    "        # Execute the query and convert results to a DataFrame\n",
    "        dependency_df = session.sql(dependency_query).to_pandas()\n",
    "\n",
    "        # Convert the result to a list of dictionaries and add schema, database info\n",
    "        dependencies = dependency_df.to_dict('records')\n",
    "\n",
    "        # Store the dependencies in the dictionary with the view name as the key\n",
    "        dependency_dict[view_name] = dependencies\n",
    "    \n",
    "    return dependency_dict\n",
    "\n",
    "def get_clustering_keys(session, table_name, schema_name, database_name):\n",
    "    \"\"\"\n",
    "    Retrieves the clustering key information for a given table.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param table_name: Name of the table\n",
    "    :param schema_name: Name of the schema\n",
    "    :param database_name: Name of the database\n",
    "    :return: Clustering key or None if no clustering key exists\n",
    "    \"\"\"\n",
    "    # Query to get the clustering key from the INFORMATION_SCHEMA for the given table\n",
    "    clustering_key_query = f\"\"\"\n",
    "    SELECT CLUSTERING_KEY\n",
    "    FROM {database_name}.INFORMATION_SCHEMA.TABLES\n",
    "    WHERE TABLE_NAME = '{table_name}'\n",
    "    AND TABLE_SCHEMA = '{schema_name}';\n",
    "    \"\"\"\n",
    "    \n",
    "    # Execute the query and fetch the result\n",
    "    clustering_key_df = session.sql(clustering_key_query).to_pandas()\n",
    "\n",
    "    if not clustering_key_df.empty:\n",
    "        return clustering_key_df['CLUSTERING_KEY'].iloc[0]\n",
    "    else:\n",
    "        return None\n",
    "\n",
    "# Function to clean up the clustering key\n",
    "def clean_clustering_key(clustering_key):\n",
    "    if pd.isna(clustering_key):\n",
    "        return None\n",
    "    # Use regex to extract the content inside parentheses\n",
    "    match = re.search(r'\\((.*?)\\)', clustering_key)\n",
    "    if match:\n",
    "        return match.group(1)  # Return the key inside parentheses\n",
    "    return clustering_key\n",
    "\n",
    "def check_dynamic_table_type(session, df_secure_views_with_shared_info, database_name):\n",
    "    \"\"\"\n",
    "    Creates dynamic tables for the views in df_secure_views_with_shared_info and retrieves the refresh mode for each table.\n",
    "    \n",
    "    :param session: Snowflake session object\n",
    "    :param df_secure_views_with_shared_info: DataFrame of view names and their shared information\n",
    "    :param database_name: Name of the database where dynamic tables will be created\n",
    "    \n",
    "    :return: DataFrame containing view names and their corresponding refresh_mode values\n",
    "    \"\"\"\n",
    "    results = []\n",
    "\n",
    "    # Iterate through each row in the DataFrame\n",
    "    for index, row in df_secure_views_with_shared_info.iterrows():\n",
    "        view_name = row['TABLE_NAME']\n",
    "        schema_name = row['TABLE_SCHEMA']\n",
    "        view_ddl = row['VIEW_DEFINITION']  # Extract the DDL directly from the DataFrame\n",
    "        \n",
    "        if not view_ddl:\n",
    "            print(f\"Skipping view {view_name}: No DDL found.\")\n",
    "            continue\n",
    "\n",
    "        # Extract the select statement from the DDL\n",
    "        select_statement = extract_query_from_view_ddl(view_ddl)\n",
    "\n",
    "        # Create the dynamic table SQL\n",
    "        create_table_sql = f\"\"\"\n",
    "        CREATE OR REPLACE DYNAMIC TABLE {database_name}.{schema_name}.test_dynamic_table\n",
    "        COMMENT = '{{\"origin\":\"sf_sit\",\"name\":\"dt_secure_view_conversion_task\",\"version\":{{\"major\":1, \"minor\":1}},\"attributes\":{{\"table\":\"{view_name}\", \"type\":\"dynamictable\"}}}}'\n",
    "        TARGET_LAG = DOWNSTREAM \n",
    "        WAREHOUSE = 'COHORT_BUILDER_LOAD_WH'\n",
    "        AS\n",
    "        {select_statement};\n",
    "        \"\"\"\n",
    "        # Execute the create table SQL\n",
    "        session.sql(create_table_sql).collect()\n",
    "\n",
    "        # Query to check the refresh mode\n",
    "        check_refresh_mode_query = f\"SHOW DYNAMIC TABLES LIKE 'test_dynamic_table' IN SCHEMA {database_name}.{schema_name};\"\n",
    "        check_refresh_mode = session.sql(check_refresh_mode_query).collect()\n",
    "\n",
    "        # Extract the refresh_mode and refresh_mode_reason column values\n",
    "        if check_refresh_mode:\n",
    "            refresh_mode_value = check_refresh_mode[0]['refresh_mode']\n",
    "            refresh_mode_reason = check_refresh_mode[0]['refresh_mode_reason']\n",
    "        else:\n",
    "            refresh_mode_value = None\n",
    "            refresh_mode_reason = None\n",
    "\n",
    "        # Store the result as a tuple of view name and refresh mode\n",
    "        results.append((view_name, refresh_mode_value, refresh_mode_reason))\n",
    "\n",
    "        # Drop the dynamic table after checking\n",
    "        session.sql(f\"DROP DYNAMIC TABLE {database_name}.{schema_name}.test_dynamic_table\").collect()\n",
    "\n",
    "    # Convert the results to a DataFrame\n",
    "    df_results = pd.DataFrame(results, columns=[\"view_name\", \"refresh_mode\", \"refresh_mode_reason\"])\n",
    "\n",
    "    return df_results\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e2491637-87ca-424d-b926-b4d47b6b211a",
   "metadata": {
    "collapsed": false,
    "name": "Step_1_Label"
   },
   "source": [
    "# 1: Identify Secured Views in Database\n",
    "\n",
    "In this section, you will select the **database** from which you would like to convert the secured views to dynamic tables. \n",
    "\n",
    "Use the following code cell to select the database for migration. It will browse through the database and return all the secure views available.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9b5b2f95-9f98-4b20-b753-f36316788ac6",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "Identify_Secured_Views"
   },
   "outputs": [],
   "source": [
    "database_name = 'MARKETING_DATA'\n",
    "\n",
    "df_shares = get_shares(session)\n",
    "    \n",
    "# Step 2: Filter shares related to the selected database\n",
    "df_shares_filtered = filter_shares_by_database(df_shares, database_name)\n",
    "\n",
    "# Step 3: Get all secure views in the selected database\n",
    "df_secure_views = get_secure_views(session, database_name)\n",
    "\n",
    "# Step 4: Get a list of views that are a part of shares in this database\n",
    "shared_views = get_all_shared_views(session, df_shares_filtered)\n",
    "\n",
    "# Step 5: Get a list of views that are secure and shared.\n",
    "df_secure_views_with_shared_info = check_if_views_are_shared(df_secure_views, shared_views)\n",
    "\n",
    "df_secure_views_with_shared_info"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ed1b0c83-f8b6-4c9e-a378-3565a6caf867",
   "metadata": {
    "collapsed": false,
    "name": "Step_2_Label_1"
   },
   "source": [
    "# 2: Extract Object Dependencies\n",
    "\n",
    "This code performs the following tasks:\n",
    "\n",
    "1. **Dependency Dictionary Creation**: \n",
    "   - The `create_dependency_dict` function creates a dictionary of dependencies for each shared view in a Snowflake database, querying the `SNOWFLAKE.ACCOUNT_USAGE.OBJECT_DEPENDENCIES` table. It extracts referenced object domain, name, schema, and catalog information for each view.\n",
    "\n",
    "Finally, the `dependency_analysis` stores the output of the shared view dependency analysis.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "be864aa6-5168-42ad-81fb-2639b43a7dd0",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "list_views_and_dependencies"
   },
   "outputs": [],
   "source": [
    "ddl_statements = {}\n",
    "view_dependencies = {}\n",
    "\n",
    "shared_views = df_secure_views_with_shared_info[df_secure_views['IS_SHARED'] == True]\n",
    "\n",
    "dependency_analysis = create_dependency_dict(session, shared_views, database_name)\n",
    "\n",
    "dependency_analysis\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dbb1b3fe-4d92-49e6-9ad7-45bea86cdf7a",
   "metadata": {
    "collapsed": false,
    "name": "Step_2_Label_2"
   },
   "source": [
    "### Building a DataFrame with View Dependencies and Clustering Keys\n",
    "\n",
    "This code processes the dependency analysis from Snowflake and constructs a detailed DataFrame containing information about each view's referenced objects, including clustering keys when applicable. The process involves the following steps:\n",
    "\n",
    "1. **DataFrame Initialization**:\n",
    "   - A new DataFrame `dependency_with_clusters_df` is created with columns: `VIEW_NAME`, `REFERENCED_OBJECT_DOMAIN`, `REFERENCED_OBJECT_NAME`, `REFERENCED_SCHEMA`, `REFERENCED_DATABASE`, and `CLUSTERING_KEY`.\n",
    "\n",
    "2. **Iterating Through Dependencies**:\n",
    "   - For each view in `dependency_analysis`, the referenced objects (like tables, schemas, databases) are extracted and stored.\n",
    "   \n",
    "3. **Clustering Key Retrieval**:\n",
    "   - For each referenced object of domain type `'TABLE'`, the clustering key is retrieved by querying the `INFORMATION_SCHEMA.TABLES` table of the corresponding database.\n",
    "   \n",
    "4. **Appending Results to DataFrame**:\n",
    "   - A new row is created for each referenced object with its corresponding clustering key (if available) and appended to the DataFrame using `pd.concat()`.\n",
    "\n",
    "The final DataFrame (`dependency_with_clusters_df`) consolidates the view dependencies along with the clustering keys, providing a comprehensive reference for further analysis.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3c03f9ea-c5db-402f-8039-24f75ed73612",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "list_dependencies_cluster"
   },
   "outputs": [],
   "source": [
    "columns = ['VIEW_NAME', 'REFERENCED_OBJECT_DOMAIN', 'REFERENCED_OBJECT_NAME', 'REFERENCED_SCHEMA', 'REFERENCED_DATABASE', 'CLUSTERING_KEY']\n",
    "dependency_with_clusters_df = pd.DataFrame(columns=columns)\n",
    "\n",
    "# Loop through the dictionary\n",
    "for view_name, dependencies in dependency_analysis.items():\n",
    "    for dependency in dependencies:\n",
    "        referenced_object_name = dependency['REFERENCED_OBJECT_NAME']\n",
    "        referenced_schema = dependency['REFERENCED_SCHEMA']\n",
    "        referenced_database = dependency['REFERENCED_DATABASE']\n",
    "        referenced_domain = dependency['REFERENCED_OBJECT_DOMAIN']\n",
    "        \n",
    "        clustering_key = None\n",
    "\n",
    "        # If the domain is 'TABLE', fetch the clustering key\n",
    "        if referenced_domain == 'TABLE':\n",
    "            # Query to get the clustering key (assuming you have a session object and function for this)\n",
    "            clustering_key_query = f\"\"\"\n",
    "            SELECT CLUSTERING_KEY\n",
    "            FROM {referenced_database}.INFORMATION_SCHEMA.TABLES\n",
    "            WHERE TABLE_NAME = '{referenced_object_name}'\n",
    "            AND TABLE_SCHEMA = '{referenced_schema}';\n",
    "            \"\"\"\n",
    "            \n",
    "            # Execute the query to retrieve the clustering key\n",
    "            clustering_key_df = session.sql(clustering_key_query).to_pandas()\n",
    "            \n",
    "            if not clustering_key_df.empty:\n",
    "                clustering_key = clustering_key_df['CLUSTERING_KEY'].iloc[0]\n",
    "        \n",
    "        # Create a new DataFrame row to append\n",
    "        new_row = pd.DataFrame([{\n",
    "            'VIEW_NAME': view_name,\n",
    "            'REFERENCED_OBJECT_DOMAIN': referenced_domain,\n",
    "            'REFERENCED_OBJECT_NAME': referenced_object_name,\n",
    "            'REFERENCED_SCHEMA': referenced_schema,\n",
    "            'REFERENCED_DATABASE': referenced_database,\n",
    "            'CLUSTERING_KEY': clustering_key\n",
    "        }])\n",
    "        \n",
    "        # Use pd.concat() to append the new row to the DataFrame\n",
    "        dependency_with_clusters_df = pd.concat([dependency_with_clusters_df, new_row], ignore_index=True)\n",
    "\n",
    "dependency_with_clusters_df['CLUSTERING_KEY'] = dependency_with_clusters_df['CLUSTERING_KEY'].apply(clean_clustering_key)\n",
    "dependency_with_clusters_df\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "27234ad0-eb28-4880-806a-84622f3dfc8f",
   "metadata": {
    "collapsed": false,
    "name": "Step_3_Label_1"
   },
   "source": [
    "# 3: Qualify Secured Views for conversion\n",
    "\n",
    "### Checking Dynamic Table Type and Refresh Mode in Snowflake\n",
    "\n",
    "This code defines the `check_dynamic_table_type` function, which performs the following tasks:\n",
    "\n",
    "1. **Purpose**:\n",
    "   - The function checks the refresh mode of dynamic tables created from secure views in Snowflake. It processes each view from the `df_secure_views_with_shared_info` DataFrame, creates a temporary dynamic table, retrieves the `refresh_mode`, and then drops the table.\n",
    "\n",
    "2. **Steps Involved**:\n",
    "   - **View Iteration**: Iterates over the rows of `df_secure_views_with_shared_info` to extract the view name, schema, and DDL (View Definition).\n",
    "   - **Select Statement Extraction**: Extracts the `SELECT` query from the view's DDL.\n",
    "   - **Dynamic Table Creation**: Creates a temporary dynamic table in Snowflake using the `CREATE OR REPLACE DYNAMIC TABLE` command.\n",
    "   - **Refresh Mode Check**: Queries the `SHOW DYNAMIC TABLES` command to check the `refresh_mode` and `refresh_mode_reason` for the created dynamic table.\n",
    "   - **Table Dropping**: Drops the dynamic table after retrieving the necessary information.\n",
    "\n",
    "3. **Output**:\n",
    "   - Returns a DataFrame containing the view names, their associated `refresh_mode`, and the reason for the refresh mode (`refresh_mode_reason`).\n",
    "\n",
    "This function helps to verify how dynamic tables created from specific views behave with respect to refresh mechanisms in Snowflake.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5f346477-2d75-4f1b-bd6b-d8aa29d9bdef",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "list_refresh_types"
   },
   "outputs": [],
   "source": [
    "check_dynamic_table_type(session, df_secure_views_with_shared_info, database_name)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "944a958c-4d9f-4c6e-9b6d-24f03d483d4f",
   "metadata": {
    "collapsed": false,
    "name": "Step_3_Label_2"
   },
   "source": [
    "### Select Secured Views for convertions\n",
    "\n",
    "Add the names of the secured views you want to filter out from the dynamic table conversion ddl that is generated in the next step"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c413a58-9369-4165-a0f4-c0b6980041ea",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "list_final_views_to_convert"
   },
   "outputs": [],
   "source": [
    "filter_list = [] \n",
    "#['ANALYTICS.VIEW_FAN_PROFILE_NONUS_TICKET_PURCHASERS']\n",
    "\n",
    "filtered_dict = {key: value for key, value in dependency_analysis.items() if key not in filter_list}\n",
    "\n",
    "filtered_dict.keys()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6972f2e5-7a5b-476d-9404-a899032ed77d",
   "metadata": {
    "collapsed": false,
    "name": "Dynamic_table_DDL"
   },
   "source": [
    "## Dynamic table DDL\n",
    "\n",
    "Please follow the steps below to run the function that generates dynamic tables. Before proceeding, ensure that all necessary clusters are reviewed. This will help optimize performance and avoid potential resource bottlenecks.\n",
    "\n",
    "### Important Notes:\n",
    "\n",
    "**Full Refresh Mode:** For any dynamic table where a full refresh is required, the function has automatically set the `REFRESH_MODE` to `FULL`. This decision is based on the complexity of the underlying query.\n",
    "\n",
    "**Overriding the Refresh Mode:**\n",
    "    If you would like to use an incremental refresh instead of a full refresh, you can override this behavior by re-creating the dynamic table with `REFRESH_MODE=INCREMENTAL`.\n",
    "    We recommend thoroughly reviewing Snowflake's guidelines on Dynamic Table Performance before making this adjustment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f165728-2a8a-4f04-95c9-2ac1af0bbd6d",
   "metadata": {
    "codeCollapsed": false,
    "collapsed": false,
    "language": "python",
    "name": "return_DDL"
   },
   "outputs": [],
   "source": [
    "get_dynamic_table_ddl_for_views(df_secure_views_with_shared_info,filtered_dict, database_name, target_lag = '10 minutes', refresh_mode = 'AUTO')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Streamlit Notebook",
   "name": "streamlit"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}