[C++] The write node options shouldn't require a schema

[asfimport]

The schema should be the output schema of the input node.

Reporter: Weston Pace / @westonpace
Assignee: Vibhatha Lakmal Abeykoon / @vibhatha

PRs and other links:

• GitHub Pull Request #12310

_{Note: This issue was originally created as ARROW-15297. Please see the migration documentation for further details.}

[asfimport]

Vibhatha Lakmal Abeykoon / @vibhatha:
@westonpace  The modification looks very clear and important. But there are a few contraints assoicated with this modification. 

I am going to think out loud and please correct me if I am wrong. 

One thing is if there is metadata or an state that needs to be consistent within an execution plan, we may face some difficulties in acessing that condition because if we don't handle it per each node in the pipeline such information could be lost. 

So should there be a mechanism like an option associated with each node option to preserve that, or the best should be to decide such and handle it in each node. 

For instance, in the current implementation (before the modification), the schema is directly passed to the write node via the scanner's information. We get the scanner's projected_schema as the output_schema to the write node. But when we use a projection as the Write function is basically scan->filter->project->write, we lose that information and cannot get the schema information from the project accurately. Should we explicitly keep an option in WriteNode to preserve that or should we handle it within the Project node. 

 

[asfimport]

Weston Pace / @westonpace:
In a schema there are two sorts of metadata. There is the top level metadata (arrow::Schema::metadata) and there is field metadata (arrow::Field::metadata). I've thought this through a bit more today and hope this helps:

---

Top level metadata and the streaming execution engine do not really work well together. The primary reason for this is that the streaming execution engine is expected to process many different batches/files of data.

If every batch of data has different top-level metadata then we would need some way to synthesize that. If every batch has the exact same top level metadata then we aren't really providing much value. So it is easier to err on the side of ignoring the top-level metadata.

At the moment we do not have many concrete use cases for top-level metadata. The only one I can think of is the pandas index which we store so we can safely round trip the files. This information would not be valid after we process a dataset regardless of what the plan is. That is because we don't currently guarantee the order of the data. So if the data was reordered at all then the index would be invalid. In addition, we might be reading multiple files. If each file has its own independent pandas index then how do we combine those?

In the future someday we may want to use the pandas index in our execution engine. For example we could materialize it into a full column and pass it through. Or we could introduce a new style of exec batch column shape which is a continuous range. We could use the metadata at scan time to properly create the exec batches. I still don't think we would end up keeping the top-level metadata as-is in the resulting output (e.g. what if we had multiple files).

---

Now let's consider field metadata

Field metadata suffers from the same problem. We have many batches of data and no way to synthesize the field metadata. However, there is a valid use case for field metadata today. We need to use the field metadata to understand the logical type of the field. When we load that data from disk/RAM into memory as a record batch we translate that "field metadata" into "logical type information". It stops being metadata and starts being actual type information. The execution engine understands type information. For example, It has to make sure every input batch has the correct types for its columns.

This is exactly the same as we would have if we took a piece of top-level metadata describing a pandas index and turned it into an actual "index column" which the execution engine understands.

So, we discard most of the field metadata but we do keep the field metadata that is used for logical types. This isn't "keeping metadata" as much as it is "consuming metadata" at read time. We then "keep type information" throughout the plan.

---

That brings us back to this PR. Maybe, if we are writing a dataset, the user wants to attach some kind of custom metadata to the dataset before they write it. So, in retrospect, I think it is probably ok for there to be an optional KeyValueMetadata option in the write node for the top-level metadata. I don't think we need to require the user provide it. I also don't think it is valid to ask the user for the schema. The schema going into the write node is beyond the control of the user. They can't decide that some column should be a string column for example if that column is not already a string column. That is what the project node is for.

It's also possible, in the future, that the user might want to calculate the top-level and field metadata for each batch that comes into the write node. For example, maybe they want to store a histogram of a column in the field metadata. Or maybe they want to store a boolean "is_sorted" flag in the field metadata. Or maybe they want to store an "authors" value in the top level metadata based on the value of one of the columns.

That is something I think we can tackle in future PRs.

[asfimport]

Vibhatha Lakmal Abeykoon / @vibhatha:
@westonpace  This make sense and it is very clear on how to manage the metadata. The purpose is very clear and it's granularity makes it harder to maintain so persistently. I also agree in future PR's we can think more about this. Although I would like to bring attention some of the test cases and assumptions we have made, I am not 100% sure if these must be changed with this change. 

I wanted to clearly understand what are the adverse side effects of causing this change. I observed a few test cases failing and it seems they are making an assumption that the metadata is preserved. One main goal for me was to determine the vision and objectives for metadata before moving further. 

Looking into this case:

arrow/python/pyarrow/tests/test_dataset.py

Line 4160 in bcf3d3e

def test_write_dataset_schema_metadata(tempdir):

Here we assume the write operation preserves the metadata. I think this was still true, because in the writer we maintained that by providing the output schema directly from the scanner not by the previous node. In the case of writing the previous node is the project where the metadata is not handled as far as I understood. 

I made the change which was very trivial to get the input schema to writer from the previous node, but this test case is failing and similar others. I assume the assumption made here is the metadata should be intact when a write operation is done. 

[asfimport]

Weston Pace / @westonpace:
Thanks for the clarification. I agree we should proceed with caution if it means invalidating an existing test case. In the test case we are writing a dataset from a table. It makes sense in that situation that we could preserve the metadata.

However, we can also write a dataset from a scanner. We might group multiple tables and input files into a single output file. What should the metadata of that output file be?

If we add the optional KeyValueMetadata as an option to FileSystemDatasetWriteOptions (in addition to the sink node) then I think we can make the test pass. We could even default to passing the table schema's metadata when the user calls write_table or write_dataset with a single table (e.g. not using a scanner as input).

CC @jorisvandenbossche for a second opinion on all of this.

[asfimport]

Vibhatha Lakmal Abeykoon / @vibhatha:
cc  @westonpace  Can we close this issue since we updated this in PR: #12721 ?

 

[asfimport]

Weston Pace / @westonpace:
Yes, I think so.