My problem was: I had a number of databases generated in different machines and I wanted to query them as if they were one, using the database the data came from as a field while querying and while showing results. The databases are SQLite3 files, generated using SQLAlchemy in a Python program.
I solved this by using SQLAlchemy, which was good because I could use the same ORM mapping that the program used. I noticed that the
Horizontal Sharding SQLAlchemy extension would fit well the problem, although not perfectly. I had to make some changes in some classes of this extension, and now it works fine.
It was possible to filter the data using database as a criteria, but I couldn't get the database information from each line of a query result. I made a simple patch to SQLAlchemy, which wasn't likely to be introduced in the distribution, but worked for me, and sent it to
its bug tracker. The change was included in SQLAlchemy in a very different fashion, as expected, but since I'm using the released version from SQLAlchemy, I kept on using my version of the patch. I don't want to do direct changes in SQLAlchemy source code, so I made the change in my program:
class ShardedSessionShardId(ShardedSession):
def __init__(self, *args, **kwargs):
super(ShardedSessionShardId, self).__init__(*args, **kwargs)
self._query_cls = ShardedQueryShardId
class ShardedQueryShardId(ShardedQuery):
def _execute_and_instances(self, context):
if self._shard_id is not None:
result = self.session.connection(
mapper=self._mapper_zero(),
shard_id=self._shard_id).execute(context.statement, self._params)
news = list(self.instances(result, context))
for new in news:
new.shard_id = self._shard_id
return iter(news)
else:
partial = []
for shard_id in self.query_chooser(self):
result = self.session.connection(
mapper=self._mapper_zero(),
shard_id=shard_id).execute(context.statement, self._params)
news = list(self.instances(result, context))
for new in news:
new.shard_id = shard_id
partial = partial + news
# if some kind of in memory 'sorting'
# were done, this is where it would happen
return iter(partial)
create_session = sessionmaker(class_=ShardedSessionShardId)
Another problem is that I had to make each result be included in the query, even if two results from different DBs have the same primary key. I achieved this by changing two classes: WeakInstanceDict, and Mapper. For using the new WeakInstanceDict, I had again to change the ShardedSession variation:
class WeakInstanceDictNoIdentity(WeakInstanceDict):
def add(self, state):
# if state.key in self:
# if dict.__getitem__(self, state.key) is not state:
# raise AssertionError("A conflicting state is already "
# "present in the identity map for key %r"
# % (state.key, ))
# else:
dict.__setitem__(self, state.key, state)
self._manage_incoming_state(state)
class ShardedSessionShardId(ShardedSession):
def __init__(self, *args, **kwargs):
super(ShardedSessionShardId, self).__init__(*args, **kwargs)
self._query_cls = ShardedQueryShardId
self._identity_cls = WeakInstanceDictNoIdentity
self.identity_map = self._identity_cls()
To start using the new Mapper, I simply replaced each call to mapper with MapperNoIdentity:
class MapperNoIdentity(Mapper):
def _instance_processor(self, context, path, adapter,
polymorphic_from=None, extension=None,
only_load_props=None, refresh_state=None,
polymorphic_discriminator=None):
"""Produce a mapper level row processor callable
which processes rows into mapped instances."""
pk_cols = self.primary_key
if polymorphic_from or refresh_state:
polymorphic_on = None
else:
if polymorphic_discriminator is not None:
polymorphic_on = polymorphic_discriminator
else:
polymorphic_on = self.polymorphic_on
polymorphic_instances = util.PopulateDict(
self._configure_subclass_mapper(
context, path, adapter)
)
version_id_col = self.version_id_col
if adapter:
pk_cols = [adapter.columns[c] for c in pk_cols]
if polymorphic_on is not None:
polymorphic_on = adapter.columns[polymorphic_on]
if version_id_col is not None:
version_id_col = adapter.columns[version_id_col]
identity_class = self._identity_class
def identity_key(row):
return identity_class, tuple([row[column] for column in pk_cols])
new_populators = []
existing_populators = []
load_path = context.query._current_path + path
def populate_state(state, dict_, row, isnew, only_load_props):
if isnew:
if context.propagate_options:
state.load_options = context.propagate_options
if state.load_options:
state.load_path = load_path
if not new_populators:
new_populators[:], existing_populators[:] = \
self._populators(context, path, row,
adapter)
if isnew:
populators = new_populators
else:
populators = existing_populators
if only_load_props:
populators = [p for p in populators
if p[0] in only_load_props]
for key, populator in populators:
populator(state, dict_, row)
session_identity_map = context.session.identity_map
if not extension:
extension = self.extension
translate_row = extension.get('translate_row', None)
create_instance = extension.get('create_instance', None)
populate_instance = extension.get('populate_instance', None)
append_result = extension.get('append_result', None)
populate_existing = context.populate_existing or self.always_refresh
if self.allow_partial_pks:
is_not_primary_key = _none_set.issuperset
else:
is_not_primary_key = _none_set.issubset
def _instance(row, result):
if translate_row:
ret = translate_row(self, context, row)
if ret is not EXT_CONTINUE:
row = ret
if polymorphic_on is not None:
discriminator = row[polymorphic_on]
if discriminator is not None:
_instance = polymorphic_instances[discriminator]
if _instance:
return _instance(row, result)
# determine identity key
if refresh_state:
identitykey = refresh_state.key
if identitykey is None:
# super-rare condition; a refresh is being called
# on a non-instance-key instance; this is meant to only
# occur within a flush()
identitykey = self._identity_key_from_state(refresh_state)
else:
identitykey = identity_key(row)
# instance = session_identity_map.get(identitykey)
# if instance is not None:
# state = attributes.instance_state(instance)
# dict_ = attributes.instance_dict(instance)
# isnew = state.runid != context.runid
# currentload = not isnew
# loaded_instance = False
# if not currentload and \
# version_id_col is not None and \
# context.version_check and \
# self._get_state_attr_by_column(
# state,
# dict_,
# self.version_id_col) != \
# row[version_id_col]:
# raise orm_exc.ConcurrentModificationError(
# "Instance '%s' version of %s does not match %s"
# % (state_str(state),
# self._get_state_attr_by_column(
# state, dict_,
# self.version_id_col),
# row[version_id_col]))
# elif refresh_state:
if refresh_state:
# out of band refresh_state detected (i.e. its not in the
# session.identity_map) honor it anyway. this can happen
# if a _get() occurs within save_obj(), such as
# when eager_defaults is True.
state = refresh_state
instance = state.obj()
dict_ = attributes.instance_dict(instance)
isnew = state.runid != context.runid
currentload = True
loaded_instance = False
else:
# check for non-NULL values in the primary key columns,
# else no entity is returned for the row
if is_not_primary_key(identitykey[1]):
return None
isnew = True
currentload = True
loaded_instance = True
if create_instance:
instance = create_instance(self,
context,
row, self.class_)
if instance is EXT_CONTINUE:
instance = self.class_manager.new_instance()
else:
manager = attributes.manager_of_class(
instance.__class__)
# TODO: if manager is None, raise a friendly error
# about returning instances of unmapped types
manager.setup_instance(instance)
else:
instance = self.class_manager.new_instance()
dict_ = attributes.instance_dict(instance)
state = attributes.instance_state(instance)
state.key = identitykey
# manually adding instance to session. for a complete add,
# session._finalize_loaded() must be called.
state.session_id = context.session.hash_key
session_identity_map.add(state)
if currentload or populate_existing:
if isnew:
state.runid = context.runid
context.progress[state] = dict_
if not populate_instance or \
populate_instance(self, context, row, instance,
only_load_props=only_load_props,
instancekey=identitykey, isnew=isnew) is \
EXT_CONTINUE:
populate_state(state, dict_, row, isnew, only_load_props)
else:
# populate attributes on non-loading instances which have
# been expired
# TODO: apply eager loads to un-lazy loaded collections ?
if state in context.partials or state.unloaded:
if state in context.partials:
isnew = False
(d_, attrs) = context.partials[state]
else:
isnew = True
attrs = state.unloaded
# allow query.instances to commit the subset of attrs
context.partials[state] = (dict_, attrs)
if not populate_instance or \
populate_instance(self, context, row, instance,
only_load_props=attrs,
instancekey=identitykey, isnew=isnew) is \
EXT_CONTINUE:
populate_state(state, dict_, row, isnew, attrs)
if loaded_instance:
state._run_on_load(instance)
if result is not None and \
(not append_result or
append_result(self, context, row, instance,
result, instancekey=identitykey,
isnew=isnew)
is EXT_CONTINUE):
result.append(instance)
return instance
return _instance
I had to include some auxiliary definitions to make the rewrites work:
_none_set = frozenset([None])
_runid = 1L
_id_lock = util.threading.Lock()
def _new_runid():
global _runid
_id_lock.acquire()
try:
_runid += 1
return _runid
finally:
_id_lock.release()
It would be good to be able to set these identity requirements as a parameter.
My last problem was selecting more than one database to search. set
shard only
worked for one, so I created a new field in query, called shards, and checked
for it on querychooser:
def query_chooser(query):
try:
return query.shards
except AttributeError:
pass
return tcs.keys()
So, when I want to look only in a list of shards, I set this field. I'm aware
that this is not a recommended python idiom, but, well, it works fine.
The question was recently asked, How do I perform crash recovery on a MySQL
master/slave database cluster? Because replication is async, a transaction
that has been committed to the master may not be able to leave the master
before the crash happens.
The short answer is, you don t. Asynchronous replication, by it s very
nature, is prone to these sorts of issues, and MySQL doesn t make life
easier in general by using query replay.
Some of the issues aren t massive: for instance, if a slave crashes, the
master will keep replication logs for the slave until it comes back up (as
long as the slave comes back before the master decides some of the replogs
are too old and starts deleting them, of course), so as long as you can live
without a master for a while, you can recover and continue along your merry
way.
Promoting a slave to be a master, on the other hand, is a situation frought
with peril. Here, you ve just got to hope hope that your chosen slave is
consistent with the master s state at death (because it s async, you have no
guarantees about that), and that all the other slaves have the same ideas
about what counts as the current state of the data. If your newly-promoted
slave managed to apply an update that another slave didn t, that slave will
be out of sync with reality (until such time as the now-dead master comes
back and replays all those queries hahahaha). To even guarantee that
any slaves have a consistent view of the data as compared to the new master,
you ve got to rereplicate everything from the new master, because MySQL s
where are you up to? counter is master-specific.
I ve heard people who have to deal with this sort of thing these days say
that the Maatkit tools are handy for dealing with
various issues of this type (monitoring, repairing the database when it gets
inconsistent).
However, I prefer to keep right the hell away from MySQL replication
altogether, after a year of having to gently coax an insane MySQL tree
replication setup to do the right thing for any extended period, and
watching a group of DBAs at Engine Yard go slowly insane.
My preferred alternatives are:
The problem
Some programs require root privileges for part of their work: most
notably Debian building tools such as dpkg-buildpackage or pdebuild. So, several command can be used to obtain real or fake root
privileges: su, sudo, fakeroot, etc. Among them, su is quite distinct,
because it only takes a quoted command to pass it to a shell. That is
why its interface is completely incompatible with the others':
I've been very quiet on here for a few months, and the reason is (of course) because I've been doing other things. No surprise there, I guess!
What I've been doing, first of all, is writing 
(This is a part of a series of posts on my recent experience with Windows. See 
Since a few months, I have been playing around with 
Here's why:
I'm afraid that in the end I failed the challenge. As the last minutes of Saturday night ticked down, I was still mired in the minutiae of C++ Run-Time Type Identification (RTTI.) Sunday was my 12th wedding anniversary and I know from past experience that this is one day when it is advisable to step away from the computer. So I have to concede defeat for now. But I only actually spent about 48 hours on this project this week so if you really think about it, I still have 120 hours available. That's what I choose to believe anyway.
The galling thing is in theory I know how this is supposed to work. 
As some of you know, yesterday we started the two weeks of activities for the 
Piespy is a bot that sits in the IRC channel logging everything that any given user "says" directly to another one, and graphs it. Not much to say here, only that it generates a very good (and practically real-time) view of what happens on the channel. For this particular channel, and during EDUSOL, you can 
Arrived at Manila yesterday. At least in the parts we passed through
(Pasay, Makati, and Malate,) everything seems back to normal, with
almost no sign of the 