feat: metadata cache provider and datafusion trait impls #25566

hiltontj · 2024-11-18T21:46:46Z

This adds the MetaCacheProvider for managing metadata caches in the influxdb3 instance. This includes APIs to create caches through the WAL as well as from a catalog on initialization, to write data into the managed caches, and to query data out of them.

The query side is fairly involved, relying on Datafusion's TableFunctionImpl and TableProvider traits to make querying the cache using a user-defined table function (UDTF) possible.

The predicate code was modified to only support two kinds of predicates: IN and NOT IN, which simplifies the code, and maps nicely with the DataFusion LiteralGuarantee which we leverage to derive the predicates from the incoming queries.

MetaCacheExec, a custom ExecutionPlan implementation was added specifically for the metadata cache that can report the predicates that are pushed down to the cache during query planning/execution.

A big set of tests was added to to check that queries are working, and that predicates are being pushed down properly.

Additional Notes

This PR moved the code in the meta_cache module of the influxdb3_cache crate around so that the primary mod.rs just contains tests, and then has the following additional modules:
- cache.rs: (existing with some changes) core cache implementation - I left a couple comments on where the main changes were made to that code
- provider.rs: (new) contains code for the new MetaCacheProvider
- table_function.rs: (new) contains code for DataFusion trait implementations
The feature still needs to be accessible through the API, so that caches can be created, deleted, and viewed, but that will be done in follow-on issues as part of Epic: Metadata Cache #25539.

hiltontj · 2024-11-21T20:04:07Z

influxdb3_cache/src/meta_cache/cache.rs

+    /// Compare the configuration of a given cache, producing a helpful error message if they differ
+    pub(crate) fn compare_config(&self, other: &Self) -> Result<(), anyhow::Error> {
+        if self.max_cardinality != other.max_cardinality {
+            bail!(
+                "incompatible `max_cardinality`, expected {}, got {}",
+                self.max_cardinality,
+                other.max_cardinality
+            );
+        }
+        if self.max_age != other.max_age {
+            bail!(
+                "incompatible `max_age`, expected {}, got {}",
+                self.max_age.as_secs(),
+                other.max_age.as_secs()
+            )
+        }
+        if self.column_ids != other.column_ids {
+            bail!(
+                "incompatible column id selection, expected {}, got {}",
+                self.column_ids
+                    .iter()
+                    .map(|id| id.to_string())
+                    .collect::<Vec<_>>()
+                    .join(","),
+                other
+                    .column_ids
+                    .iter()
+                    .map(|id| id.to_string())
+                    .collect::<Vec<_>>()
+                    .join(","),
+            )
+        }
+
+        Ok(())
+    }


This is newly added to the core cache code, as it is used on cache creation in the MetaCacheProvider to give friendly errors if users attempt to overwrite an existing cache.

hiltontj · 2024-11-21T20:05:50Z

influxdb3_cache/src/meta_cache/cache.rs

+/// A predicate that can be applied when gathering [`RecordBatch`]es from a [`MetaCache`]
+///
+/// This is intended to be derived from a set of filter expressions in Datafusion by analyzing
+/// them with a `LiteralGuarantee`.
+///
+/// This uses a `BTreeSet` to store the values so that they are iterated over in sorted order.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) enum Predicate {
+    In(BTreeSet<Value>),
+    NotIn(BTreeSet<Value>),
+}


This type was changed from the previous enum that also contained Eq and NotEq variants. Those were no longer needed with how predicates are pushed down from DataFusion (see the table_function.rs module)

This adds the MetaDataCacheProvider for managing metadata caches in the influxdb3 instance. This includes APIs to create caches through the WAL as well as from a catalog on initialization, to write data into the managed caches, and to query data out of them. The query side is fairly involved, relying on Datafusion's TableFunctionImpl and TableProvider traits to make querying the cache using a user-defined table function (UDTF) possible. The predicate code was modified to only support two kinds of predicates: IN and NOT IN, which simplifies the code, and maps nicely with the DataFusion LiteralGuarantee which we leverage to derive the predicates from the incoming queries. A custom ExecutionPlan implementation was added specifically for the metadata cache that can report the predicates that are pushed down to the cache during query planning/execution. A big set of tests was added to to check that queries are working, and that predicates are being pushed down properly.

hiltontj · 2024-11-22T02:49:13Z

influxdb3_cache/src/meta_cache/mod.rs

-                    "| us-west | d    |",
-                    "+---------+------+",
-                ],
-            },


All of these test cases were removed because the Eq and NotEq variants were removed from the Predicate type.

praveen-influx

Nice PR - comments were very handy to follow the changes. I learnt few DF tricks as well when going through this. 🙏 Just left few queries for my understanding.

praveen-influx · 2024-11-22T10:59:03Z

influxdb3_catalog/src/catalog.rs

@@ -691,6 +692,7 @@ pub struct TableDefinition {
    pub column_map: BiHashMap<ColumnId, Arc<str>>,
    pub series_key: Option<Vec<ColumnId>>,
    pub last_caches: HashMap<Arc<str>, LastCacheDefinition>,
+    pub meta_caches: HashMap<Arc<str>, MetaCacheDefinition>,


I think we need to clear out the metacache when table or db is deleted? Maybe address in one of the follow up PRs

I will see if I can address this now, since I did have to merge in your delete work anyways, it shouldn't be too hard to make this additional change.

praveen-influx · 2024-11-22T11:30:42Z

influxdb3_cache/src/meta_cache/table_function.rs

+                    write!(f, " predicates=[")?;
+                    let mut p_iter = predicates.iter();
+                    while let Some((col_id, predicate)) = p_iter.next() {
+                        write!(f, "[{col_id} {predicate}]")?;


I like we have a custom display to plug into explain output. Maybe we could swap column name for id here?

predicates=[[0 IN (us-east)], [1 IN (a,b)]]

probably reads better (imo) when its

predicates=[[region IN (us-east)], [host IN (a,b)]]

Just an UX thing, when there's many columns involved this might be a little difficult to understand. If col_id is useful in explain, then we could add that info here like region$0 or some variation of it?

Good call, I think I could do this by passing forward the Arc<TableDefinition> to the MetaCacheExec so it only pulls the column names when needed, e.g., for the EXPLAIN output. The table definition is already there so should not be too tricky, and yeah - given that part of the motivation for this whole execution plan implementation is to make it readable to the operator so I will get this in.

praveen-influx · 2024-11-22T13:35:40Z

influxdb3_cache/src/meta_cache/table_function.rs

+    // discard those predicates and have DataFusion handle the filtering.
+    //
+    // This is a conservative approach; it may be that we can combine multiple literal guarantees on
+    // a single column, but thusfar, from testing in the parent module, this does not seem necessary.


This whole thing is quite interesting, thanks for summarizing it. I'd need to go through this closely to understand it fully but I think I generally the get the idea.

praveen-influx · 2024-11-22T14:10:57Z

influxdb3_cache/src/meta_cache/cache.rs

+/// whose values hold the last seen time as an [`i64`] of each value, and an optional reference to
+/// the node in the next level of the tree.
+#[derive(Debug, Default)]
+struct Node(BTreeMap<Value, (i64, Option<Node>)>);


Just for my clarity, does this mean each level in the tree will correspond to a column (in turn a BTreeMap)? If that's the case,

does the filtering take advantage of the fact that it doesn't need to scan subsequent levels if the query doesn't include it, for eg. select region from metadata_cache('cpu') will not require host to be scanned. It sounds like something we'll do, but I couldn't figure out where we short circuit it.

do we need to support out of band queries, like select host from metadata_cache('cpu') (note: there's no where clause for region)?

does the filtering take advantage of the fact that it doesn't need to scan subsequent levels if the query doesn't include it, for eg. select region from metadata_cache('cpu') will not require host to be scanned. It sounds like something we'll do, but I couldn't figure out where we short circuit it.

It does not. That would be "projection pushdown" which we could accomplish by handling the projection arg passed in the TableProvider::scan method. I did not attempt that yet. I will add an issue to tackle this since it is definitely doable but trickier than I want to handle on this PR and some requirements might need refining.

do we need to support out of band queries, like select host from metadata_cache('cpu') (note: there's no where clause for region)?

It does handle this now, and there is a test that does this, but it is doing a full scan of the cache hierarchy, and as I note in that test, then DataFusion does the filtering/projection - so this is not optimal.

We might be able to improve handling of this if we tackle projection pushdown, but will need to decide some things, e.g., if the output should be sorted or not.

praveen-influx · 2024-11-22T14:12:11Z

influxdb3_cache/src/meta_cache/mod.rs

+                    "| us-east | b    |",
+                    "+---------+------+",
+                ],
+                explain_contains: "MetaCacheExec: predicates=[[0 IN (us-east)], [1 IN (a,b)]] inner=MemoryExec: partitions=1, partition_sizes=[1]",


These explain_contains are really neat 👍

Yes this is super helpful for testing. I originally had the entire EXPLAIN output in the test but unfortunately it was flaky since some things differ in CI 😔. There might be some helpers in datafusion that we could use to normalize the EXPLAIN output, will need to dig for that.

hiltontj · 2024-11-22T15:56:29Z

@praveen-influx - on further thought, I have made some forward progress on subsequent issues so I want to get this merged. I opened issues to address the feedback you gave:

hiltontj force-pushed the hiltontj/meta-cache-write-path branch 4 times, most recently from 5d159c9 to 583f69e Compare November 21, 2024 20:02

hiltontj commented Nov 21, 2024

View reviewed changes

hiltontj force-pushed the hiltontj/meta-cache-write-path branch 2 times, most recently from 44d0ec2 to 9dc7cc8 Compare November 21, 2024 20:35

hiltontj requested review from pauldix, mgattozzi and praveen-influx and removed request for pauldix and mgattozzi November 21, 2024 20:35

hiltontj self-assigned this Nov 21, 2024

hiltontj added the v3 label Nov 21, 2024

hiltontj marked this pull request as ready for review November 21, 2024 20:35

hiltontj force-pushed the hiltontj/meta-cache-write-path branch from 9dc7cc8 to 088df39 Compare November 21, 2024 20:44

Merge branch 'main' into hiltontj/meta-cache-write-path

20c74a7

hiltontj mentioned this pull request Nov 22, 2024

Validate that predicate pushdown is working in the last cache #25578

Open

hiltontj commented Nov 22, 2024

View reviewed changes

Merge branch 'main' into hiltontj/meta-cache-write-path

20d09a8

praveen-influx approved these changes Nov 22, 2024

View reviewed changes

This was referenced Nov 22, 2024

Display column names in MetaCacheExec EXPLAIN output #25582

Open

Delete metadata caches on database/table delete #25583

Open

hiltontj merged commit 8e23032 into main Nov 22, 2024
13 checks passed

hiltontj deleted the hiltontj/meta-cache-write-path branch November 22, 2024 15:57

hiltontj mentioned this pull request Nov 22, 2024

Additional testing on the metadata cache #25564

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: metadata cache provider and datafusion trait impls #25566

feat: metadata cache provider and datafusion trait impls #25566

hiltontj commented Nov 18, 2024 •

edited

Loading

hiltontj Nov 21, 2024 •

edited

Loading

hiltontj Nov 21, 2024

hiltontj Nov 22, 2024

praveen-influx left a comment

praveen-influx Nov 22, 2024

hiltontj Nov 22, 2024

hiltontj Nov 22, 2024

praveen-influx Nov 22, 2024

hiltontj Nov 22, 2024

hiltontj Nov 22, 2024

praveen-influx Nov 22, 2024

praveen-influx Nov 22, 2024

hiltontj Nov 22, 2024

hiltontj Nov 22, 2024

praveen-influx Nov 22, 2024

hiltontj Nov 22, 2024

hiltontj commented Nov 22, 2024

feat: metadata cache provider and datafusion trait impls #25566

feat: metadata cache provider and datafusion trait impls #25566

Conversation

hiltontj commented Nov 18, 2024 • edited Loading

Additional Notes

hiltontj Nov 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

praveen-influx left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hiltontj commented Nov 22, 2024

hiltontj commented Nov 18, 2024 •

edited

Loading

hiltontj Nov 21, 2024 •

edited

Loading