Switch to new data passing API #109

nicklan · 2024-01-27T01:53:37Z

WIP of the "big bang" switch to the new data passing API. Note this is still based on when I named things "EngineClient" and not "EngineInterface" so that term is used throughout. I will rebase to main and rename everything once we're agreed on the major pieces.

This implements:

The traits for the API as discussed
- including list/map
Switching the defs in lib to use it
Parsing of Metadata, Protocol, and Add
A SimpleClient which is a simple EngineClient that uses arrow but isn't async
Has the default client return EngineData
working scan
- data-skipping still uses arrow for now, we will fully remove it as a followup. See Remove last bits of Arrow from data_skipping.rs #126
Remove any mention of arrow in lib.rs

Potential things still to do:

Probably move SimpleData to ArrowData and not in the simple_client mod since it can be used by both Default and Simple client

doesn't touch DefaultTableClient yet

method is unimplemented! in default client for now

bit ugly on the expression stuff atm

ryan-johnson-databricks

Nice! We're sooo close to banishing the last bits of arrow from core kernel code. Just that one spot in data skipping that messes with selection vectors.

(bunch of unresolved comments from before, plus several new ones)

data-skipping still uses arrow for now, we can change that after #83 merges

It merged.

ryan-johnson-databricks · 2024-02-23T23:04:51Z

kernel/src/actions/mod.rs

-    if metadata.is_none() {
-        return Ok(Box::new(std::iter::empty()));
+    pub fn schema(&self) -> DeltaResult<StructType> {
+        Ok(serde_json::from_str(&self.schema_string)?)


Do we eventually need to use the engine's json parser for this?

Well, it's internal so we don't have to I don't think

Internal and a single value that won't hog MB/GB memory. Fair.

I do remember we discussed some related topic. In general the main issue here I think is that we cannot tell the engine what schema we are expecting which right now is required. IIRC the question was mainly trying to avoid serde_json as a dependency, personally I would vote for accepting as a dependency and doing it as implemented here.

Another place whas trying to parse the _last_cehckpoint file, which may also contain the schema of the checkpoint files - I think. At that time we would also have no way of telling eninge the schema of the data we want. In this case though I belive that using that schema was not eneven doen internally at databricks?

ryan-johnson-databricks · 2024-02-23T23:07:36Z

kernel/src/client/expression.rs

+            let schema = ArrowSchema::new(vec![ArrowField::new("output", arrow_type, true)]);
+            RecordBatch::try_new(Arc::new(schema), vec![array_ref])?


Out of curiosity -- do you know what code currently triggers this non-struct path?

You can tell from DataSkippingFilter::new. For instance the STATS_EXPR is a column expression, so it'll eval to just an Array, so the select_stats_evaluator will trigger this path (I think, haven't tested exactly what does but I do know the unit tests hit both paths)

kernel/src/lib.rs

kernel/src/scan/data_skipping.rs

ryan-johnson-databricks · 2024-02-23T23:30:35Z

kernel/src/simple_client/mod.rs

+
+impl EngineInterface for SimpleClient {
+    fn get_expression_handler(&self) -> Arc<dyn ExpressionHandler> {
+        unimplemented!();


This just needs us to factor out the expression handler that the default client uses?
Or is there some bigger blocker?

Nope, that's all it would take yeah. Just would prefer to do that as a follow-up

kernel/src/actions/deletion_vector.rs

ryan-johnson-databricks · 2024-02-23T23:36:09Z

kernel/src/actions/deletion_vector.rs

+
+pub(crate) fn treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
+    fn combine(high_bits: u32, low_bits: u32) -> usize {
+        ((u64::from(high_bits) << 32) | u64::from(low_bits)) as usize


Rescuing #109 (comment) from github oblivion...

this would silently lose information if sizeof(usize) < sizeof(u64); we should add a static assertion to make it fail at compile time instead.

(even if we move it from kernel to engine, the same issue would remain in our default clients)

Yep, I've filed #124 since we'd need to add a dependency

kernel/src/scan/mod.rs

kernel/src/scan/data_skipping.rs

ryan-johnson-databricks

Nice!

ryan-johnson-databricks · 2024-02-27T03:58:26Z

kernel/src/engine_data.rs

+/// Any type that an engine wants to return as "data" needs to implement this trait. The bulk of the
+/// work is in the [`EngineData::extract`] method. See the docs for that method for more details.
+/// ```rust
+/// # use std::any::Any;


what does the # mean?

Because cargo doc actually builds the code in the comments to make sure the examples are correct, you have to import stuff. But if you don't want it to show up in the docs because it's not relevant, you can just # it. So it means the example will just show like:

TIL, thanks!

ryan-johnson-databricks · 2024-02-27T13:32:38Z

kernel/src/actions/mod.rs

-    if metadata.is_none() {
-        return Ok(Box::new(std::iter::empty()));
+    pub fn schema(&self) -> DeltaResult<StructType> {
+        Ok(serde_json::from_str(&self.schema_string)?)


Internal and a single value that won't hog MB/GB memory. Fair.

ryan-johnson-databricks · 2024-02-27T13:56:57Z

kernel/src/client/json.rs

+            return Ok(Box::new(SimpleData::new(RecordBatch::new_empty(
+                output_schema,
+            ))));


Agree that's not better.

(I'm still building intuition for when type inference works vs. doesn't, thanks for your patience)

ryan-johnson-databricks · 2024-02-27T20:42:34Z

I believe this is "mergable" (once conflicts are resolved), and we can work to fully migrate in follow-up PRs.

Conflicts are resolved now, correct?

nicklan · 2024-02-27T21:24:39Z

I believe this is "mergable" (once conflicts are resolved), and we can work to fully migrate in follow-up PRs.

Conflicts are resolved now, correct?

Yep, removed from the description. Just waiting for one more review now, I've pinged people

zachschuermann

awesome stuff nick LGTM I just left a few random nits/questions/comments

zachschuermann · 2024-03-04T16:28:13Z

kernel/src/actions/deletion_vector.rs

+use crate::{DeltaResult, Error, FileSystemClient};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DeletionVectorDescriptor {


do we want this to be pub? Could it just be pub(crate)?

I think we do want actions to be pub, since we expect external systems might use them. at the very least, all of the action stuff is pub for now, so I'll leave it for a follow-up discussion :)

zachschuermann · 2024-03-04T16:35:58Z

kernel/src/actions/deletion_vector.rs

+
+    use super::DeletionVectorDescriptor;
+
+    fn dv_relateive() -> DeletionVectorDescriptor {


nit

Suggested change

fn dv_relateive() -> DeletionVectorDescriptor {

fn dv_relative() -> DeletionVectorDescriptor {

zachschuermann · 2024-03-04T17:48:04Z

kernel/src/actions/mod.rs

-    Ok(Box::new(zipped.flatten().map(Action::Add)))
-}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Add {


at risk of being repetitive: do we want actions to be pub? I was hoping we could hide more of the implementation details instead of exposing a large API with internal details like standalone did..

oh yea just noticed below - Remove is pub(crate)

Yeah, it's a good question which I'll leave for a later discussion :)

maybe we could hide it behind the visibility crate as well?

I think right now there is a files list method exposed externally, but my hope was anyhow to make that in terms of data...

zachschuermann · 2024-03-04T17:48:43Z

kernel/src/actions/mod.rs

-        handler.parse_json(json_strings, output_schema).unwrap()
+impl Add {
+    /// Since we always want to parse multiple adds from data, we return a `Vec<Add>`
+    pub fn parse_from_data(data: &dyn EngineData) -> DeltaResult<Vec<Add>> {


awesome how these turned out :)

zachschuermann · 2024-03-04T17:50:33Z

kernel/src/actions/mod.rs

-            .collect::<Vec<_>>();
-        println!("{:?}", actions)
+impl Remove {
+    // _try_new_from_data for now, to avoid warning, probably will need at some point


Yeah, I think we'll want this at some point, so I'm just going to leave it so it's less work then. If it hangs out for a while and we end up not needing it, I will remove it.

zachschuermann · 2024-03-04T18:03:27Z

kernel/src/scan/file_stream.rs

@@ -19,6 +20,35 @@ struct LogReplayScanner {
    seen: HashSet<(String, Option<String>)>,
 }

+#[derive(Default)]
+struct AddRemoveVisitor {


is this just for visiting a mix of add/removes for replay?

yep, it just handles visiting one or the other so we only have to traverse the data once.

kernel/src/snapshot.rs

zachschuermann · 2024-03-04T18:08:49Z

kernel/src/snapshot.rs

@@ -462,14 +438,14 @@ mod tests {
        assert!(cp.is_none())
    }

-    #[test]
+    #[test_log::test]


should we have many other tests using test_log?

we probably should, this was just a really useful one for debugging so I left it in. i think we should actually just always have it, which is maybe just a matter of always importing it or something, but again, future work :)

zachschuermann · 2024-03-04T18:09:36Z

kernel/src/lib.rs

+pub mod arrow_conversion;
+
+#[cfg(feature = "simple-client")]
+pub mod simple_client;


wow awesome example/implementation of a client

Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>

roeap

Very sorry I did not get to it last friday - dropping my pending comments in case someone is interested :).

roeap · 2024-02-17T22:41:27Z

kernel/src/scan/file_stream.rs

+                Ok(adds) => Either::Left(adds.into_iter().map(Ok)),
+                Err(err) => Either::Right(std::iter::once(Err(err))),
+            }
+        }
        Err(err) => Either::Right(std::iter::once(Err(err))),


the main reason for this is that rust wants us to return exactly one specific kind of iterator for the impl Iterator<Item = DeltaResult<Add>>, but Vec::Iter and the Once iterator are different things. Eiher helps us to conveniently normalise this. Without collecting the all actions internally, I see no clear path to avoid this. This does not mean there is none though :)

roeap · 2024-02-29T17:28:31Z

kernel/src/actions/mod.rs

+    /// The time when this metadata action is created, in milliseconds since the Unix epoch
+    pub created_time: Option<i64>,
+    /// Configuration options for the metadata action
+    pub configuration: HashMap<String, Option<String>>,


This has been the situation as I found it when first joining with delta-rs, and I remember this bothering me a bunch of times, but I don't recall exactly what the reason was we could not get rid of it.

I'll try to remember, but if we can move to just String, this will allow he consuming code to avoid a bunch of None checks...

roeap · 2024-02-29T17:34:36Z

kernel/src/actions/mod.rs

-    if metadata.is_none() {
-        return Ok(Box::new(std::iter::empty()));
+    pub fn schema(&self) -> DeltaResult<StructType> {
+        Ok(serde_json::from_str(&self.schema_string)?)


I do remember we discussed some related topic. In general the main issue here I think is that we cannot tell the engine what schema we are expecting which right now is required. IIRC the question was mainly trying to avoid serde_json as a dependency, personally I would vote for accepting as a dependency and doing it as implemented here.

Another place whas trying to parse the _last_cehckpoint file, which may also contain the schema of the checkpoint files - I think. At that time we would also have no way of telling eninge the schema of the data we want. In this case though I belive that using that schema was not eneven doen internally at databricks?

roeap · 2024-03-07T03:59:46Z

kernel/src/actions/mod.rs

-    Ok(Box::new(zipped.flatten().map(Action::Add)))
-}
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Add {


maybe we could hide it behind the visibility crate as well?

I think right now there is a files list method exposed externally, but my hope was anyhow to make that in terms of data...

roeap · 2024-03-07T04:01:31Z

kernel/src/actions/mod.rs

-        let output_schema = Arc::new(log_schema().clone());
-        handler.parse_json(json_strings, output_schema).unwrap()
+impl Add {
+    /// Since we always want to parse multiple adds from data, we return a `Vec<Add>`


nit this feels more like a comment (//) then a docstring (///)?

nicklan added 22 commits January 23, 2024 17:29

rename table client -> engine client

6c9efe1

doesn't touch DefaultTableClient yet

add engine_data and trait method

ec08901

method is unimplemented! in default client for now

checkpoint, working parsing metadata/protocol

f6c1072

checkpoint, can parse metadata in snapshot

0643945

make scan tests pass (with some todos)

53c8850

All internal tests pass

32a28a9

clean up parsing a bit

d7930e6

remove old comment

0e54c24

add todo comment

438a1e2

New better way to do list/map

5ab0021

put back old test

ab89728

add dv support

e182e3d

start to think about DVs

6b3efe3

make dv.rs tests work

fd80a97

fmt and small fix

4bf5b47

add some tests back

304da21

fix "list_from"

a7393fe

remove todo

54a0c1a

allow default client to use EngineData

54a0d7f

use some unsafe magic to make more tests pass

a8e8920

fmt

33923bb

some clippy guided cleanup

82cff1b

nicklan requested review from roeap, ryan-johnson-databricks, wjones127 and zachschuermann February 1, 2024 01:21

nicklan added 4 commits February 2, 2024 16:47

switch to selection vec

dc6e963

make parse_json not refrence record batch anymore

1cff896

almost have arrow out of lib.rs

e6d5eb5

add back md test

28031fa

nicklan added 4 commits February 22, 2024 16:14

checkpoint, actually compiles. needs expressions to be put back

a2dedf5

woo, all tests working.

f5d80c3

bit ugly on the expression stuff atm

add is_empty to make clippy happy

0a9cd11

final clippy fixes

b268b4e

nicklan requested review from roeap and ryan-johnson-databricks February 23, 2024 22:16

nicklan added 5 commits February 23, 2024 14:17

import style

aab7fb8

put back test that merge inexplicably removed

8d397e6

use error constructors where strings are static

a2cb373

missed one

1a19d19

impl From for SimpleData -> RecordBatch

c130692

ryan-johnson-databricks reviewed Feb 23, 2024

View reviewed changes

bunch of doc fixes

fdd927a

nicklan mentioned this pull request Feb 26, 2024

make Scan::execute return an iterator #123

Closed

nicklan added 2 commits February 26, 2024 15:11

small comment changes

b4699c0

fmt

1c922c9

ryan-johnson-databricks approved these changes Feb 27, 2024

View reviewed changes

nicklan mentioned this pull request Feb 28, 2024

Use a proc macro for being able to derive schemas for our action structs #129

Closed

nicklan requested a review from rtyler February 28, 2024 23:42

zachschuermann approved these changes Mar 4, 2024

View reviewed changes

nicklan and others added 2 commits March 4, 2024 12:42

Update kernel/src/snapshot.rs

4a14694

Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>

fix typo

14d0aff

nicklan merged commit 4adb294 into delta-io:main Mar 4, 2024
3 checks passed

roeap reviewed Mar 7, 2024

View reviewed changes

This was referenced Mar 26, 2024

Updates for the actions module #67

Closed

Explore providing a ColumnarBatch like trait #21

Closed

Implement remaining types for ColumnarBatch #52

Closed

Settle ColumnBatch API and finish PR #41 #63

Closed

		let schema = ArrowSchema::new(vec![ArrowField::new("output", arrow_type, true)]);
		RecordBatch::try_new(Arc::new(schema), vec![array_ref])?


		use super::DeletionVectorDescriptor;

		fn dv_relateive() -> DeletionVectorDescriptor {

	fn dv_relateive() -> DeletionVectorDescriptor {
	fn dv_relative() -> DeletionVectorDescriptor {

Switch to new data passing API #109

Switch to new data passing API #109

Conversation

nicklan commented Jan 27, 2024 • edited Loading

ryan-johnson-databricks 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

ryan-johnson-databricks 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

ryan-johnson-databricks commented Feb 27, 2024

nicklan commented Feb 27, 2024

zachschuermann 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

roeap 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

nicklan commented Jan 27, 2024 •

edited

Loading