lowering consolidation to 300MB

Author: kylemann16

src/silvimetric/commands/shatter.py

@@ -30,8 +30,8 @@ def final(
     config.finished = finished
     storage.save_shatter_meta(config)
 
-    storage.consolidate(config.timestamp)
-    storage.vacuum()
+    persist(delayed(storage.consolidate)(config.timestamp))
+    compute(delayed(storage.vacuum)())
 
 
 def get_data(extents: Extents, filename: str, storage: Storage) -> pd.DataFrame:
@@ -233,7 +233,7 @@ def shatter(config: ShatterConfig) -> int:
         config.log.debug('Shattering.')
         try:
             run(leaves, config, storage)
-            storage.consolidate(config.timestamp)
+            persist(delayed(storage.consolidate)(config.timestamp))
         except Exception as e:
             final(config, storage)
             raise e

src/silvimetric/resources/storage.py

@@ -476,6 +476,7 @@ def vacuum(self):
         for vac_type in ['commits', 'fragments', 'fragment_meta']:
             c = tiledb.Config({'sm.vacuum.mode': vac_type})
             tiledb.vacuum(self.config.tdb_dir, config=c)
+        self.config.log.debug('Vacuuming complete.')
 
     def consolidate(self, timestamp) -> None:
         """
@@ -484,9 +485,11 @@ def consolidate(self, timestamp) -> None:
         time traveling.
         :param timestamp: TileDB timestamp, a tuple of start and end datetime.
         """
-        self.config.log.debug('Consolidating db.')
         for con_type in ['commits', 'fragments', 'fragment_meta']:
-            c = tiledb.Config({'sm.consolidation.mode': con_type})
+            c = tiledb.Config({
+                'sm.consolidation.mode': con_type,
+                'sm.consolidation.max_fragment_size': (300*2**20) #300MB
+            })
             tiledb.consolidate(
                 self.config.tdb_dir, timestamp=timestamp, config=c
             )