You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@madlib.apache.org by ok...@apache.org on 2022/06/22 20:31:40 UTC
[madlib] 01/03: Utilities: Reuse update plan in GroupIterationController
This is an automated email from the ASF dual-hosted git repository.
okislal pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/madlib.git
commit f37a92924d21204a83566f66e2c814ad22b88588
Author: Orhan Kislal <ok...@apache.org>
AuthorDate: Thu Jun 9 16:57:33 2022 -0400
Utilities: Reuse update plan in GroupIterationController
The group iteration controller prepares a plan and executes it during
the update phase. For some modules, this plan does not change between
iterations. With this commit, the plan gets saved and reused as needed.
Co-authored-by: Bhuvnesh Chaudhary <bc...@pivotal.io>
---
src/ports/postgres/modules/svm/svm.py_in | 1 +
.../modules/utilities/in_mem_group_control.py_in | 68 ++++++++++++----------
2 files changed, 37 insertions(+), 32 deletions(-)
diff --git a/src/ports/postgres/modules/svm/svm.py_in b/src/ports/postgres/modules/svm/svm.py_in
index 1532cb24..ceddb392 100644
--- a/src/ports/postgres/modules/svm/svm.py_in
+++ b/src/ports/postgres/modules/svm/svm.py_in
@@ -62,6 +62,7 @@ def _compute_svm(args):
it.kwargs['stepsize'] *= it.kwargs['decay_factor']
else:
it.kwargs['stepsize'] = init_stepsize / (it.iteration + 1)
+ it.update_plan = None
has_converged = it.test(
"""
{iteration} >= {max_iter}
diff --git a/src/ports/postgres/modules/utilities/in_mem_group_control.py_in b/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
index 68834c04..31f7f374 100644
--- a/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
+++ b/src/ports/postgres/modules/utilities/in_mem_group_control.py_in
@@ -240,6 +240,7 @@ class GroupIterationController:
self.finished_states = state_factory(self.is_state_type_bytea8)
self.group_param = self._init_group_param()
+ self.update_plan = None
def _init_group_param(self):
_grp_key = ("array_to_string(ARRAY[{grouping_str}], ',')"
@@ -549,40 +550,43 @@ class GroupIterationController:
self.iteration = self.iteration + 1
group_param = self.group_param
- run_sql = """
- SELECT
- {_grp_key} AS {col_grp_key},
- {grouping_col},
- {iteration} AS {col_grp_iteration},
- ({newState}) AS {col_grp_state}
- FROM (
- SELECT *,
- array_to_string(ARRAY[{grouping_str}], ',') AS {col_grp_key}
- FROM {rel_source}
- ) AS {as_rel_source}
- JOIN ( {select_rel_state} ) AS {rel_state}
- {using_str}
- JOIN ( {select_n_tuples} ) AS _rel_n_tuples
- {using_str}
- {groupby_str}
- """.format(
- newState=newState,
- iteration=self.iteration,
- using_str=group_param.using_str,
- groupby_str=group_param.groupby_str,
- _grp_key=group_param.grp_key,
- select_rel_state=group_param.select_rel_state,
- select_n_tuples=group_param.select_n_tuples,
- **self.kwargs)
- update_plan = plpy.prepare(run_sql,
- ["text[]", group_param.grouped_state_type,
- "text[]", "integer[]"])
- res_tuples = plpy.execute(update_plan, [self.new_states.keys,
- self.new_states.values,
- self.grp_to_n_tuples.keys(),
- self.grp_to_n_tuples.values()])
+ if self.update_plan is None:
+ run_sql = """
+ SELECT
+ {_grp_key} AS {col_grp_key},
+ {grouping_col},
+ ({newState}) AS {col_grp_state}
+ FROM (
+ SELECT *,
+ array_to_string(ARRAY[{grouping_str}], ',') AS {col_grp_key}
+ FROM {rel_source}
+ ) AS {as_rel_source}
+ JOIN ( {select_rel_state} ) AS {rel_state}
+ {using_str}
+ JOIN ( {select_n_tuples} ) AS _rel_n_tuples
+ {using_str}
+ {groupby_str}
+ """.format(
+ newState=newState,
+ iteration=self.iteration,
+ using_str=group_param.using_str,
+ groupby_str=group_param.groupby_str,
+ _grp_key=group_param.grp_key,
+ select_rel_state=group_param.select_rel_state,
+ select_n_tuples=group_param.select_n_tuples,
+ **self.kwargs)
+
+ self.update_plan = plpy.prepare(run_sql,
+ ["text[]", group_param.grouped_state_type,
+ "text[]", "integer[]"])
+ res_tuples = plpy.execute(self.update_plan,
+ [self.new_states.keys,
+ self.new_states.values,
+ self.grp_to_n_tuples.keys(),
+ self.grp_to_n_tuples.values()])
+ res_tuples[0][self.kwargs['col_grp_iteration']] = self.iteration
col_grp_state = self.kwargs['col_grp_state']
col_grp_key = self.kwargs['col_grp_key']
self.old_states.sync_from(self.new_states)