provide a new add_cubes() migration function for cases where the new cubes are linked together by new relations
In this case, we need to add all new cubes at once.
"""Defines the diferent querier steps usable in plans.
FIXME : this code needs refactoring. Some problems :
* get data from the parent plan, the latest step, temporary table...
* each step has is own members (this is not necessarily bad, but a bit messy
for now)
:organization: Logilab
:copyright: 2003-2009 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
"""
__docformat__ = "restructuredtext en"
from rql.nodes import VariableRef, Variable, Function
from cubicweb.server.ssplanner import (LimitOffsetMixIn, Step, OneFetchStep,
varmap_test_repr, offset_result)
AGGR_TRANSFORMS = {'COUNT':'SUM', 'MIN':'MIN', 'MAX':'MAX', 'SUM': 'SUM'}
def remove_clauses(union, keepgroup):
clauses = []
for select in union.children:
if keepgroup:
having, orderby = select.having, select.orderby
select.having, select.orderby = None, None
clauses.append( (having, orderby) )
else:
groupby, having, orderby = select.groupby, select.having, select.orderby
select.groupby, select.having, select.orderby = None, None, None
clauses.append( (groupby, having, orderby) )
return clauses
def restore_clauses(union, keepgroup, clauses):
for i, select in enumerate(union.children):
if keepgroup:
select.having, select.orderby = clauses[i]
else:
select.groupby, select.having, select.orderby = clauses[i]
class FetchStep(OneFetchStep):
"""step consisting in fetching data from sources, and storing result in
a temporary table
"""
def __init__(self, plan, union, sources, table, keepgroup, inputmap=None):
OneFetchStep.__init__(self, plan, union, sources)
# temporary table to store step result
self.table = table
# should groupby clause be kept or not
self.keepgroup = keepgroup
# variables mapping to use as input
self.inputmap = inputmap
# output variable mapping
srqlst = union.children[0] # sample select node
# add additional information to the output mapping
self.outputmap = plan.init_temp_table(table, srqlst.selection,
srqlst.solutions[0])
for vref in srqlst.selection:
if not isinstance(vref, VariableRef):
continue
var = vref.variable
if var.stinfo['attrvars']:
for lhsvar, rtype in var.stinfo['attrvars']:
if lhsvar.name in srqlst.defined_vars:
key = '%s.%s' % (lhsvar.name, rtype)
self.outputmap[key] = self.outputmap[var.name]
else:
rschema = self.plan.schema.rschema
for rel in var.stinfo['rhsrelations']:
if rschema(rel.r_type).inlined:
lhsvar = rel.children[0]
if lhsvar.name in srqlst.defined_vars:
key = '%s.%s' % (lhsvar.name, rel.r_type)
self.outputmap[key] = self.outputmap[var.name]
def execute(self):
"""execute this step"""
self.execute_children()
plan = self.plan
plan.create_temp_table(self.table)
union = self.union
# XXX 2.5 use "with"
clauses = remove_clauses(union, self.keepgroup)
for source in self.sources:
source.flying_insert(self.table, plan.session, union, plan.args,
self.inputmap)
restore_clauses(union, self.keepgroup, clauses)
def mytest_repr(self):
"""return a representation of this step suitable for test"""
clauses = remove_clauses(self.union, self.keepgroup)
try:
inputmap = varmap_test_repr(self.inputmap, self.plan.tablesinorder)
outputmap = varmap_test_repr(self.outputmap, self.plan.tablesinorder)
except AttributeError:
inputmap = self.inputmap
outputmap = self.outputmap
try:
return (self.__class__.__name__,
sorted((r.as_string(kwargs=self.plan.args), r.solutions)
for r in self.union.children),
sorted(self.sources), inputmap, outputmap)
finally:
restore_clauses(self.union, self.keepgroup, clauses)
class AggrStep(LimitOffsetMixIn, Step):
"""step consisting in making aggregat from temporary data in the system
source
"""
def __init__(self, plan, selection, select, table, outputtable=None):
Step.__init__(self, plan)
# original selection
self.selection = selection
# original Select RQL tree
self.select = select
# table where are located temporary results
self.table = table
# optional table where to write results
self.outputtable = outputtable
if outputtable is not None:
plan.init_temp_table(outputtable, selection, select.solutions[0])
#self.inputmap = inputmap
def mytest_repr(self):
"""return a representation of this step suitable for test"""
sel = self.select.selection
restr = self.select.where
self.select.selection = self.selection
self.select.where = None
rql = self.select.as_string(kwargs=self.plan.args)
self.select.selection = sel
self.select.where = restr
try:
# rely on a monkey patch (cf unittest_querier)
table = self.plan.tablesinorder[self.table]
outputtable = self.outputtable and self.plan.tablesinorder[self.outputtable]
except AttributeError:
# not monkey patched
table = self.table
outputtable = self.outputtable
return (self.__class__.__name__, rql, self.limit, self.offset, table,
outputtable)
def execute(self):
"""execute this step"""
self.execute_children()
self.inputmap = inputmap = self.children[-1].outputmap
# get the select clause
clause = []
for i, term in enumerate(self.selection):
try:
var_name = inputmap[term.as_string()]
except KeyError:
var_name = 'C%s' % i
if isinstance(term, Function):
# we have to translate some aggregat function
# (for instance COUNT -> SUM)
orig_name = term.name
try:
term.name = AGGR_TRANSFORMS[term.name]
# backup and reduce children
orig_children = term.children
term.children = [VariableRef(Variable(var_name))]
clause.append(term.accept(self))
# restaure the tree XXX necessary?
term.name = orig_name
term.children = orig_children
except KeyError:
clause.append(var_name)
else:
clause.append(var_name)
for vref in term.iget_nodes(VariableRef):
inputmap[vref.name] = var_name
# XXX handle distinct with non selected sort term
if self.select.distinct:
sql = ['SELECT DISTINCT %s' % ', '.join(clause)]
else:
sql = ['SELECT %s' % ', '.join(clause)]
sql.append("FROM %s" % self.table)
# get the group/having clauses
if self.select.groupby:
clause = [inputmap[var.name] for var in self.select.groupby]
grouped = set(var.name for var in self.select.groupby)
sql.append('GROUP BY %s' % ', '.join(clause))
else:
grouped = None
if self.select.having:
clause = [term.accept(self) for term in self.select.having]
sql.append('HAVING %s' % ', '.join(clause))
# get the orderby clause
if self.select.orderby:
clause = []
for sortterm in self.select.orderby:
sqlterm = sortterm.term.accept(self)
if sortterm.asc:
clause.append(sqlterm)
else:
clause.append('%s DESC' % sqlterm)
if grouped is not None:
for vref in sortterm.iget_nodes(VariableRef):
if not vref.name in grouped:
sql[-1] += ', ' + self.inputmap[vref.name]
grouped.add(vref.name)
sql.append('ORDER BY %s' % ', '.join(clause))
if self.limit:
sql.append('LIMIT %s' % self.limit)
if self.offset:
sql.append('OFFSET %s' % self.offset)
#print 'DATA', plan.sqlexec('SELECT * FROM %s' % self.table, None)
sql = ' '.join(sql)
if self.outputtable:
self.plan.create_temp_table(self.outputtable)
sql = 'INSERT INTO %s %s' % (self.outputtable, sql)
return self.plan.sqlexec(sql, self.plan.args)
def visit_function(self, function):
"""generate SQL name for a function"""
return '%s(%s)' % (function.name,
','.join(c.accept(self) for c in function.children))
def visit_variableref(self, variableref):
"""get the sql name for a variable reference"""
try:
return self.inputmap[variableref.name]
except KeyError: # XXX duh? explain
return variableref.variable.name
def visit_constant(self, constant):
"""generate SQL name for a constant"""
assert constant.type == 'Int'
return str(constant.value)
class UnionStep(LimitOffsetMixIn, Step):
"""union results of child in-memory steps (e.g. OneFetchStep / AggrStep)"""
def execute(self):
"""execute this step"""
result = []
limit = olimit = self.limit
offset = self.offset
assert offset != 0
if offset is not None:
limit = limit + offset
for step in self.children:
if limit is not None:
if offset is None:
limit = olimit - len(result)
step.set_limit_offset(limit, None)
result_ = step.execute()
if offset is not None:
offset, result_ = offset_result(offset, result_)
result += result_
if limit is not None:
if len(result) >= olimit:
return result[:olimit]
return result
def mytest_repr(self):
"""return a representation of this step suitable for test"""
return (self.__class__.__name__, self.limit, self.offset)
class IntersectStep(UnionStep):
"""return intersection of results of child in-memory steps (e.g. OneFetchStep / AggrStep)"""
def execute(self):
"""execute this step"""
result = set()
for step in self.children:
result &= frozenset(step.execute())
result = list(result)
if self.offset:
result = result[self.offset:]
if self.limit:
result = result[:self.limit]
return result
class UnionFetchStep(Step):
"""union results of child steps using temporary tables (e.g. FetchStep)"""
def execute(self):
"""execute this step"""
self.execute_children()
__all__ = ('FetchStep', 'AggrStep', 'UnionStep', 'UnionFetchStep', 'IntersectStep')