Merge pull request #34 from vainglorygame/partitioning

Partitioning

1 files changed, 53 insertions, 18 deletions

diff --git a/api.py b/api.py
index d178828..aafe6a5 100644
--- a/api.py
+++ b/api.py
@@ -42,6 +42,22 @@ class Apigrabber(object):
                 (id SERIAL, start_date TIMESTAMP,
                 end_date TIMESTAMP, finished BOOL, region TEXT)
             """)
+        # create master tables
+        for objecttype in ["match", "roster", "participant",
+                           "team", "player"]:
+            await con.execute("""
+                CREATE TABLE IF NOT EXISTS """ + objecttype + """ (
+                    id TEXT PRIMARY KEY NOT NULL,
+                    type TEXT NOT NULL,
+                    attributes JSONB,
+                    relationships JSONB)
+                """)
+            # create region partitions
+            # TODO use CHECK for regions once every object has a shardId
+            for region in self.regions:
+                await con.execute("CREATE TABLE IF NOT EXISTS " +
+                        objecttype + "_" + region +
+                        " (id TEXT PRIMARY KEY) INHERITS (" + objecttype + ")")
 
         # create past zombie job that marks the last data to fetch
         async with con.transaction():
@@ -55,21 +71,36 @@ class Apigrabber(object):
                 ON CONFLICT DO NOTHING;
             """, json.dumps([{"region": r} for r in self.regions]))
 
-    async def _db_insert(self, con, objects):
+    async def _db_insert(self, con, objects, ddate, region):
         """Insert a list of API response objects into respective tables."""
+        day = ddate.strftime("%Y_%m_%d")
+
+        def table(objtype):
+            """Return the partition the object belongs in."""
+            return (objtype + "_" + region +
+                    ("_" + day
+                     if objtype != "player"
+                     and objtype != "team"
+                     else ""))
+
         objectmap = {}
-        for o in objects:
-            try:
-                objectmap[o["type"]].append(o)
-            except KeyError:
-                objectmap[o["type"]] = [o]
-                await con.execute("""
-                    CREATE TABLE IF NOT EXISTS """ + o["type"] + """ (
-                        id TEXT PRIMARY KEY NOT NULL,
-                        type TEXT NOT NULL,
-                        attributes JSONB,
-                        relationships JSONB)
-                    """)
+        async with con.transaction():  # create savepoint
+            for o in objects:
+                try:
+                    objectmap[o["type"]].append(o)
+                except KeyError:
+                    objectmap[o["type"]] = [o]
+                    # create a partition for the day
+                    # players are not partitioned by day
+                    try:
+                        await con.execute("CREATE TABLE IF NOT EXISTS " +
+                                              table(o["type"]) +
+                                          " (id TEXT PRIMARY KEY) INHERITS (" +
+                                              o["type"] + "_" + region +
+                                          ")")
+                    except asyncpg.DuplicateTableError:
+                        # ninja'd by another worker
+                        pass
 
         for otype, objs in objectmap.items():
             async with con.transaction():  # create savepoint
@@ -83,7 +114,7 @@ class Apigrabber(object):
                 # recent. TODO always keep the update condition in line
                 # with the data that is returned by the API.
                 await con.execute("""
-                    INSERT INTO """ + otype + """ AS j
+                    INSERT INTO """ + table(otype) + """ AS j
                         SELECT DISTINCT ON(id) * FROM
                         JSONB_TO_RECORDSET($1::JSONB)
                         AS jsn(id TEXT, type TEXT, attributes JSONB, relationships JSONB)
@@ -117,7 +148,9 @@ class Apigrabber(object):
 
                 logging.info("%s: (%s) received %s data objects",
                              region, jobid, len(matches))
-                await self._db_insert(con, matches)
+                # TODO to be more precise, jobs that return date over midnight
+                # should be split so they insert in the according table
+                await self._db_insert(con, matches, jobstart, region)
                 logging.info("%s: (%s) inserted",
                              region, jobid)
 
@@ -149,10 +182,11 @@ class Apigrabber(object):
                         WHERE start_date-previous_end>INTERVAL '0'  -- get gaps
                         ORDER BY start_date DESC LIMIT 1
                         """, region, default_diff)
-                        if row_res == None:
+                        if row_res is None:
                             logging.warn("%s: no jobs available. idling.", region)
                             await asyncio.sleep(60)  # a minute TODO make this smarter
                             asyncio.ensure_future(self.crawl_region(region))
+                            return
 
                         jobdate, delta_minutes = row_res
                         delta = datetime.timedelta(minutes=delta_minutes)
@@ -214,8 +248,8 @@ class Apigrabber(object):
     async def setup(self):
         async with self._pool.acquire() as con:
             await self._db_setup(con)
-            ## clean up after force quit (TODO - disabled for dev)
-            #await con.execute("DELETE FROM crawljobs WHERE finished=false")
+            # clean up after force quit (TODO - disabled for dev)
+            await con.execute("DELETE FROM crawljobs WHERE finished=false")
 
 
 async def startup():
@@ -234,3 +268,4 @@ logging.basicConfig(level=logging.DEBUG)
 loop = asyncio.get_event_loop()
 loop.run_until_complete(startup())
 loop.run_forever()
+ 
\ No newline at end of file