models/sms_model_reconfiguration_costs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Special Case Model: Segment-based Model with Sorting Dependencies (SMS) with Reconfiguration Costs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# import pyomo\n",
    "from pyomo.environ import *\n",
    "from pyomo.opt import SolverFactory\n",
    "\n",
    "# import pandas\n",
    "import pandas as pd\n",
    "\n",
    "# import helper\n",
    "from helper.print import print_result\n",
    "from helper.export import export_config"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load benchmark, system calibration and workload data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# performance measurements of the benchmark queries\n",
    "df_perf = pd.read_csv('../data/benchmark/runtimes.csv')\n",
    "# memory consumption of segments\n",
    "df_memory = pd.read_csv('../data/benchmark/memory_consumption.csv')\n",
    "# memory consumption of indexes\n",
    "df_memory_index = pd.read_csv('../data/benchmark/memory_consumption_index.csv')\n",
    "\n",
    "# calibration data for hyrise \n",
    "df_poslist_scan_penalty = pd.read_csv('../data/calibration/poslist_scan_penalty.csv')\n",
    "# calibration data for storage devices\n",
    "df_storage_penalty = pd.read_csv('../data/calibration/storage_penalty.csv')\n",
    "\n",
    "# workload definition \n",
    "df_workload = pd.read_csv('../data/workloads/workload_1/workload.csv')\n",
    "# chunk access statistics \n",
    "df_chunk_access = pd.read_csv('../data/workloads/workload_1/chunk_access.csv')\n",
    "\n",
    "# Apply indexes on pandas dataframes\n",
    "df_perf.set_index(['ORDER_BY', 'ENCODING', 'SCAN_COLUMN', 'SELECTIVITY', 'INDEX', 'SCAN_TYPE'], inplace=True)\n",
    "df_memory.set_index(['ORDER_BY', 'ENCODING', 'COLUMN', 'CHUNK_ID'], inplace=True)\n",
    "df_memory_index.set_index(['ORDER_BY', 'ENCODING', 'COLUMN', 'CHUNK_ID'], inplace=True)\n",
    "df_poslist_scan_penalty.set_index(['ENCODING'], inplace=True)\n",
    "df_storage_penalty.set_index(['STORAGE', 'ENCODING', 'INDEX'], inplace=True)\n",
    "df_chunk_access.set_index(['QUERY_ID', 'CHUNK'], inplace=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Base Config"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "CONFIG_PATH = '../data/base_config/'\n",
    "\n",
    "def load_config(file_name, path=None):\n",
    "    if path is None:\n",
    "        path = CONFIG_PATH\n",
    "    df_base_config = pd.read_csv(path + file_name + '.csv')\n",
    "    df_base_config.set_index(['CHUNK', 'COLUMN'], inplace=True)\n",
    "    return df_base_config\n",
    "\n",
    "def to_df_config(model):\n",
    "    lines = []\n",
    "    sort_columns = {} \n",
    "    \n",
    "    config = model.X\n",
    "    for item in config:\n",
    "        if round(config[item].value) == 1.0:\n",
    "            lines += [item]\n",
    "            if item[3] == 1:\n",
    "                sort_columns[item[0]] = item[1]\n",
    "    \n",
    "    lines = set_sorted_column_for_segment(lines, sort_columns)\n",
    "    df = pd.DataFrame.from_records(lines, columns=[\"CHUNK\", \"COLUMN\", \"ENCODING\", \"SORT\", \"INDEX\", \"STORAGE\"])\n",
    "    df.set_index(['CHUNK', 'COLUMN'], inplace=True)\n",
    "    return df\n",
    "\n",
    "def change_base_config(df, model):\n",
    "    BASE_CONFIG = df\n",
    "    model.R.reconstruct()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# In Hyrise, scan operations performed on the pos list are significantly slower than isolated \n",
    "# executed scan operations. To consider this in our cost estimation, we introduce an encoding-specific \n",
    "# parameter, which can be measured during the model calibration. \n",
    "def scan_order_penalty(scan_order, encoding):\n",
    "    return 1 if (scan_order == 0) else df_poslist_scan_penalty.loc[(encoding)]['PENALTY']\n",
    "\n",
    "# storage penalties\n",
    "def storage_penalty(encoding, index, storage):\n",
    "    return df_storage_penalty.loc[(storage, encoding, index)]['PENALTY']\n",
    "\n",
    "# set index value to 0 for all scans with a scan order value >= 1\n",
    "# indexes can only be used to speed up the first scan\n",
    "def index_value(scan, i):    \n",
    "    return i if scan['SCAN_ORDER'] == 0 else 0\n",
    "\n",
    "# the measured execution time of an isolated scan operation with the given column configuration\n",
    "def performance(s, e, o, i):\n",
    "    scan = df_workload.iloc[s]\n",
    "    return df_perf.loc[(o, e, scan['SCAN_COLUMN'], scan['SELECTIVITY'], index_value(scan, i), scan['SCAN_TYPE'])]['TIME']\n",
    "    \n",
    "# updates the storage budgets (the amount of storage units can not be modified)\n",
    "def update_storage_budgets(storage_budgets, model):\n",
    "    model.SB.reconstruct(storage_budgets)\n",
    "    model.MemoryBudgetConstraint.reconstruct()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ConcreteModel()\n",
    "\n",
    "# memory budget of the different storage devices    \n",
    "STORAGE_BUDGET = {0:4_000_000_000, 1:3_000_000_000}\n",
    "\n",
    "# base configuration, which is currently applied\n",
    "BASE_CONFIG = load_config('base_config')\n",
    "\n",
    "# modification costs per segment\n",
    "ALPHA = 500_000_000\n",
    "COSTS_ENCODING_CHANGE = 2\n",
    "COSTS_CREATE_INDEX = 1.5\n",
    "COSTS_SORTING_CHANGE = 2.5\n",
    "COSTS_STORAGE_CHANGE = 1\n",
    "\n",
    "# set of order configurations \n",
    "O = df_perf.index.levels[0].unique()\n",
    "\n",
    "# set of encodings\n",
    "E = df_perf.index.levels[1].unique()\n",
    "\n",
    "# set of columns\n",
    "N = df_perf.index.levels[2].unique()\n",
    "\n",
    "# set of chunks\n",
    "M = df_memory.index.levels[3].unique()\n",
    "\n",
    "# set of scan operations defined in the workload  \n",
    "model.S = Set(initialize=range(0, df_workload.shape[0]))\n",
    "\n",
    "# set of storage devices\n",
    "model.B = Set(initialize=range(0, len(STORAGE_BUDGET)))\n",
    "\n",
    "# storage budget per storage device\n",
    "model.SB = Param(model.B, within=NonNegativeIntegers, initialize=STORAGE_BUDGET, mutable=True)\n",
    "\n",
    "# discrete variable for the indexing configuration of a segment\n",
    "model.I = Set(initialize=[0,1], within=Binary)\n",
    "\n",
    "# decision variable to describe selected combination of options \n",
    "model.X = Var(M, N, E, O, model.I, model.B, within=Binary) \n",
    "\n",
    "# decision variable to describe the selected index and sort configuration for chunk m\n",
    "model.Y = Var(M, O, within=Binary)\n",
    "\n",
    "# decision variable to describe the selected encoding configuration for a segment m,n \n",
    "model.Z = Var(M, N, E, model.I, model.B, within=Binary)\n",
    "\n",
    "# returns the approximated costs of single segment scan based on the overall column scan costs\n",
    "# proportional costs depending on the number of accessed chunks\n",
    "def segment_access_init(model, m, n, s):\n",
    "    q = df_workload.iloc[s]['QUERY_ID']\n",
    " \n",
    "    if df_workload.iloc[s]['SCAN_COLUMN'] != n or df_chunk_access.loc[(q, m)]['ACCESSED'] == 0:\n",
    "        return 0\n",
    "    return (1/df_chunk_access.loc[(q, slice(None))]['ACCESSED'].sum())\n",
    "model.A = Param(M, N, model.S, within=NonNegativeReals, initialize=segment_access_init)\n",
    "\n",
    "# Runtime performance for column per encoding, sorting, indexing, and selectivity \n",
    "def performance_init(model, m, n, e, o, i, b):\n",
    "    return sum(df_workload.iloc[s]['FREQUENCY'] * \\\n",
    "               performance(s, e, o, i) * \\\n",
    "               model.A[m, n, s] * \\\n",
    "               df_workload.iloc[s]['SCAN_FACTOR'] * \\\n",
    "               scan_order_penalty(df_workload.iloc[s]['SCAN_ORDER'], e) * \\\n",
    "               storage_penalty(e, i, b)\n",
    "               for s in model.S)\n",
    "model.C = Param(M, N, E, O, model.I, model.B, within=NonNegativeReals, initialize=performance_init, mutable=True)\n",
    "\n",
    "# memory consumption of a segment for a given encoding ordering, and sorting configuration\n",
    "def memory_init(model, m, n, e, o, i):\n",
    "    if i == 0:\n",
    "        return df_memory.loc[(o,e,n,m)]['SIZE_IN_BYTES']\n",
    "    else:\n",
    "        return df_memory.loc[(o,e,n,m)]['SIZE_IN_BYTES'] + df_memory_index.loc[(o,e,n,m)]['SIZE_IN_BYTES']\n",
    "model.MC = Param(M, N, E, O, model.I, within=NonNegativeReals, initialize=memory_init)\n",
    "\n",
    "# returns if a given index encoding configuration can be implemented in a database\n",
    "# for hyrise only indices on dictionary encoded columns are allowed\n",
    "def valid_index_encoding_config(model, e, i):\n",
    "    if i == 1 and e == 0:\n",
    "        return 1\n",
    "    return 0\n",
    "model.VC = Param(E, model.I, within=Binary, initialize=valid_index_encoding_config)\n",
    "\n",
    "# determine modification costs for each configuration based on applied base config\n",
    "def modification_costs_init(model, m, n, e, o, i, b):\n",
    "    mod_costs = 0\n",
    "    segment_config = BASE_CONFIG.loc[(m,n)]\n",
    "    \n",
    "    if segment_config['ENCODING'] !=  e:\n",
    "        mod_costs += COSTS_ENCODING_CHANGE\n",
    "    if segment_config['SORT'] ==  n and o != segment_config['SORT']:\n",
    "        mod_costs += COSTS_SORTING_CHANGE\n",
    "    if segment_config['INDEX'] == 0 and i > 0:\n",
    "        mod_costs += COSTS_CREATE_INDEX\n",
    "    if segment_config['STORAGE'] != b:\n",
    "        mod_costs += COSTS_STORAGE_CHANGE\n",
    "    return mod_costs \n",
    "model.R = Param(M, N, E, O, model.I, model.B, within=NonNegativeReals, initialize=modification_costs_init, mutable=True)\n",
    "\n",
    "### Objective\n",
    "\n",
    "def runtime(m):\n",
    "    return sum(m.X[chunk_id, column_id, encoding_id, ordering_id, index_config, storage_id] * \\\n",
    "               m.C[chunk_id, column_id, encoding_id, ordering_id, index_config, storage_id]\n",
    "               for chunk_id in M\n",
    "               for column_id in N\n",
    "               for encoding_id in E\n",
    "               for ordering_id in O\n",
    "               for index_config in m.I\n",
    "               for storage_id in m.B) + ALPHA * \\\n",
    "           sum(m.X[chunk_id, column_id, encoding_id, ordering_id, index_config, storage_id] * \\\n",
    "               m.R[chunk_id, column_id, encoding_id, ordering_id, index_config, storage_id]\n",
    "               for chunk_id in M\n",
    "               for column_id in N\n",
    "               for encoding_id in E\n",
    "               for ordering_id in O\n",
    "               for index_config in m.I\n",
    "               for storage_id in m.B)\n",
    "model.Obj = Objective(rule=runtime)\n",
    "\n",
    "### Constraints \n",
    "\n",
    "# Size within budget\n",
    "def memory_budget_rule(m, b):\n",
    "    # sum up memory consumption of all \n",
    "    return sum((m.X[chunk_id, column_id, encoding_id, ordering_id, index_config, b] * \\\n",
    "                m.MC[chunk_id, column_id, encoding_id, ordering_id, index_config])\n",
    "               for chunk_id in M\n",
    "               for column_id in N\n",
    "               for encoding_id in E\n",
    "               for ordering_id in O\n",
    "               for index_config in m.I) <= m.SB[b]\n",
    "model.MemoryBudgetConstraint = Constraint(model.B, rule=memory_budget_rule)\n",
    "\n",
    "# one column sorted per chunk\n",
    "def single_sorting_column_per_chunk_rule(m, i):\n",
    "    return sum(m.Y[i, ordering_id] \n",
    "               for ordering_id in O) == 1\n",
    "model.SingleSortColumnConstraints = Constraint(M, rule=single_sorting_column_per_chunk_rule)\n",
    "\n",
    "# Exactly one encoding, sorting, and indexing configuration per segment\n",
    "def one_encoding_sorting_indexing_config_active_per_segment_rule(m, i, j):\n",
    "    return sum(m.Z[i, j, encoding_id, index_config, storage_id] \n",
    "                for encoding_id in E\n",
    "                for index_config in m.I\n",
    "                for storage_id in m.B) == 1\n",
    "model.SingleEncodingConstraints = Constraint(M, N, rule=one_encoding_sorting_indexing_config_active_per_segment_rule)\n",
    "\n",
    "# all segmets of a chunk have to be stored on the same storage medium\n",
    "# def single_storage_medium_per_chunk_rule(m, i, j):\n",
    "#     return sum(m.Z[i, 1, encoding_id, index_config, storage_id] * storage_id\n",
    "#                for encoding_id in E\n",
    "#                for index_config in m.I\n",
    "#                for storage_id in m.B) == \\\n",
    "#            sum(m.Z[i, j, encoding_id, index_config, storage_id] * storage_id\n",
    "#                for encoding_id in E\n",
    "#                for index_config in m.I\n",
    "#                for storage_id in m.B)       \n",
    "# model.SingleStorageConstraints = Constraint(M, N, rule=single_storage_medium_per_chunk_rule)\n",
    "\n",
    "# lin X = Z * Y\n",
    "def linear_x(m, j, n, e, o, i, b):\n",
    "    return m.X[j, n, e, o, i, b] >= m.Y[j, o] + m.Z[j, n, e, i, b] - 1 \n",
    "model.linearX = Constraint(M, N, E, O, model.I, model.B, rule=linear_x)\n",
    "\n",
    "def linear_y(m, j, n, e, o, i, b):\n",
    "     return m.X[j, n, e, o, i, b] <= m.Y[j, o] \n",
    "model.linearY = Constraint(M, N, E, O, model.I, model.B, rule=linear_y)\n",
    "\n",
    "def linear_z(m, j, n, e, o, i, b):\n",
    "     return m.X[j, n, e, o, i, b] <= m.Z[j, n, e, i, b] \n",
    "model.linearZ = Constraint(M, N, E, O, model.I, model.B, rule=linear_z)\n",
    "\n",
    "# Indexed columns need to have dictionary encoding\n",
    "def valid_index_configuration_rule(m, i, j, k, l, n):\n",
    "    return m.X[i, j, k, l, 1, n] <= m.VC[k, 1]\n",
    "model.IndexesOnDictColumnsConstraints = Constraint(M, N, E, O, model.B, rule=valid_index_configuration_rule)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Solving"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Solving for budget;\n",
      "  Storage: 0     Storage Size: 50000000 \n",
      "  Storage: 1     Storage Size: 75000000 \n",
      "\n",
      "Result: optimal (walltime: 63.7961s)\n",
      "Objective: 4093353240.5196285\n",
      "Memory consumption:\n",
      "  0: 49.98806403104718 MB\n",
      "  1: 38.788684 MB\n",
      "\n",
      "Storage: 0\n",
      "CHUNK               driver_id           latitude            longitude           timestamp           status              \n",
      "0                   S - RunLength       - - FoR-SIMD        - - FoR-SIMD        - - Dictionary      - - LZ4             \n",
      "1                   S - RunLength       - - FoR-SIMD        - - LZ4             - - Dictionary      - - LZ4             \n",
      "2                   - - Dictionary      S - FoR-SIMD        - - FoR-SIMD        - - Dictionary      - - LZ4             \n",
      "3                   - - Dictionary      S - FoR-SIMD        - - FoR-SIMD        - - Dictionary      - - LZ4             \n",
      "4                   - - Dictionary      S - FoR-SIMD        - - FoR-SIMD        - - Dictionary      - - LZ4             \n",
      "5                   - - Dictionary      S - FoR-SIMD        - - FoR-SIMD        - - Dictionary      - - FoR-SIMD        \n",
      "6                   \n",
      "7                   \n",
      "8                   \n",
      "9                   \n",
      "\n",
      "Storage: 1\n",
      "CHUNK               driver_id           latitude            longitude           timestamp           status              \n",
      "0                   \n",
      "1                   \n",
      "2                   \n",
      "3                   \n",
      "4                   \n",
      "5                   \n",
      "6                   - - LZ4             S - FoR-SIMD        - - LZ4             - - Dictionary      - - LZ4             \n",
      "7                   - - LZ4             S - FoR-SIMD        - - LZ4             - - Dictionary      - - LZ4             \n",
      "8                   - - LZ4             S - FoR-SIMD        - - LZ4             - - Dictionary      - - LZ4             \n",
      "9                   - I Dictionary      S - FoR-SIMD        - - LZ4             - - Dictionary      - - LZ4             \n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "solver = SolverFactory('gurobi')\n",
    "solver.options['threads'] = 16\n",
    "\n",
    "result = solver.solve(model)\n",
    "print_result(result, model, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}