Fix materialize commit

[bohou-aryn]

Replace the dataset operators with docset operators to handle
materialize success handling.

[dhruvkaliraman7]

Can we have some tests please, not obvious what the code is doing.

[eric-anderson]

I think this code is going to force execution twice.
You don't know the count unless you execute the entire pipeline. I think that self.count() will run through the standard Executor so will materialize (if that's enabled) and at least not re-do the full effort.

A more efficient implementation would be to do a streaming read and use the choose k out of n algorithm from https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle; so you store the 1000 documents, and process all of them exactly once. I'd probably just put this in as a TODO.

[eric-anderson]

Remember to add the TODO.

[eric-anderson]

I don't understand what this is for. I think what is happening is it's assuming that some other work happened to produce dataset from node, and that we return that, but I don't think this will interact correctly with our plan rewriting stuff. I think the dataset is computed earlier (and I can't quite tell exactly when it is forced to be computed. but if plan rewriting inserts some nodes, those inserts wouldn't be respected.

Probably just need to chat.

[eric-anderson]

docs = sycamore.init().read().map().map().filter().llm_generate_group().aggregate().
take_all() { actual_node = rewrite(node); return [ Document(d) for d in actual_node.execute() ] }
add a node rewriter that adds a filter step right after read (for security purposes)
The filter will be ignored because llm_generate_group will have calculated the pipeline when the filter step doesn't exist.

def aggregate(self, child):
    ds = child.execute() # this computes a ray dataset, which binds the execution tree to what is happening with child at the time of the call
    return DataSetTransplant(ds, child)

1. node tree up to aggregate
2. transform that node tree into a ray dataset
3. add a node after the initial read
4. call take_all which ignores the node added in step 3 -- security is broken here because the filter doesn't happen

child = [1,2,3,4]
dataset = map(lambda x: x + 1, child)
child[3] = 17
# dataset is unchanged

execute() function to turn a node into a ray_dataset

If node -> ray form was effectively 
  ray_dataset = ray_dataset.ray_apply(lambda: node_as_function(node))
  this would let you update node and have take effect until ray runs.
  but you couldn't insert a node.
  
but the actual code looks more like:
  node = Filter(...)
  ray_dataset = ray_dataset.filter(fn = node.fn)
  now updating node.fn has no effect

nodes = Scan Node -> Entity Extraction -> GroupBy [not a docset here] -> Aggregate.
ds = agg() # DocSet
class DocSet:
     def agg(self):
           return DocSet(Node(...))

ctx = sycamore.init()
ds1 = ctx.read()
# here is where the rewrite rule would insert the filter
ds2 = ds1.ee()
# Eric guesses adding rule here will work
gb = ds2.gb()
# Here might work
ds3 = gb.collect()
# Here won't work because collect has already called execute on ds2, so the rule is ignored.
docs = ds.take_all()

---

This is hard to figure out what it is doing.
Write a test with a rewrite rule that inserts a filter to remove one of the groups and verify the group vanishes with the rule enabled.

[eric-anderson]

Remember to add the TODO.

[eric-anderson]

# TODO: share a base class between AggregateCount and AggregateCollect. The respective execute functions are identical.