refactor(cubestore): Replace invalidate_tables_cache flag with generi… (#10683)

Introduce a generic `set_post_commit_callback` on `BatchPipe<S>` that fires after a
successful RocksDB commit on the RW-loop thread. This replaces the hard-coded
`invalidate_tables_cache` boolean flag, which was a hack that also ran before the
actual commit.

- `BatchPipe` is now generic over a store type `S` (defaults to `()`), allowing
  callbacks to receive a typed store reference
- `RocksMetaStore::write_operation` passes `&RocksMetaStore` to callbacks, giving
  them direct access to `cached_tables` and other metastore state
- Moved `cached_tables` from `RocksStore` to `RocksMetaStore` where it belongs
- Cache invalidation now only happens when the commit actually succeeds
- `RocksTable` methods are generic over `S` so they work with any `BatchPipe<S>`

Author: ovr

rust/cubestore/cubestore/src/cachestore/cache_rocksstore.rs

@@ -498,14 +498,14 @@ impl RocksCacheStore {
         f: F,
     ) -> Result<R, CubeError>
     where
-        F: for<'a> FnOnce(DbTableRef<'a>, &'a mut BatchPipe) -> Result<R, CubeError>
+        F: for<'a> FnOnce(DbTableRef<'a>, &mut BatchPipe<'a>) -> Result<R, CubeError>
             + Send
             + Sync
             + 'static,
         R: Send + Sync + 'static,
     {
         self.store
-            .write_operation_impl::<F, R>(&self.rw_loop_queue_cf, op_name, f)
+            .write_operation_impl::<F, R, ()>(&self.rw_loop_queue_cf, op_name, f, ())
             .await
     }
 

rust/cubestore/cubestore/src/lib.rs

@@ -214,6 +214,12 @@ impl From<ParserError> for CubeError {
     }
 }
 
+impl From<regex::Error> for CubeError {
+    fn from(v: regex::Error) -> Self {
+        CubeError::from_error(v)
+    }
+}
+
 impl From<ParquetError> for CubeError {
     fn from(v: ParquetError) -> Self {
         CubeError::from_error(v.to_string())