to_parquet cannot convert union struct

[jpivarski]

From @nongiga:

I'll try to do this as easy to reproduce as I can but it's a little complicated.

I have a data structure, that looks a little like this:

{
    "CaseNum": 1905256,
    "Isolates": [
      {
        "IsoNum": 5567545
        "SeqsPlate": "1_E5",
      }
     Seqplates: [
      "1_E5",
      "1_D5"
    ]
    ],
  }, 

I am loading multiple csv files and saving their content in a subfield of the structure, making a final struct that looks more like this:

{
    "CaseNum": 1905256,
    "Isolates": [
      {
        "IsoNum": 5567545
        "SeqsPlate": "1_E5",
      }
     Seqplates: [ "1_E5", "1_D5" ]
    ],
   "Pangenome":[
       "Gene": [gene1, gene2, ...]
      "Annotation": [annot1, annot2, ...]
    ]
  }, 

I did it using the following code, a bit bukly but get the job done:

datadir=os.getcwd()

# get names of csv files
dirnames=np.char.add(np.asarray(datadir+"/pangenome/", dtype=str), np.asarray(df.CaseNum,dtype=str))
filenames=np.char.add(dirnames,np.asarray('/gene_presence_absence.csv', dtype=str))

# keep track over which of the files exist
isfile=[os.path.isfile(fn) for fn in filenames]

# function to read csv files into awkwardarray
def get_pangenome_data(filename):
    roary_pa=csv.read_csv(filename)
    roary_ak=ak.from_arrow(roary_pa)
   # returns AK and the length of the array for later unflattening
    return roary_ak, len(roary_ak)

# loop through all existing csv files and get their data
pangenome_data=[ (get_pangenome_data(fn))  for fn in filenames[isfile] ]

# get line count in each file for unflattening
r_count=np.array([c for dat, c in pangenome_data])
count=np.zeros((len(ilp),),dtype=int) # all line counts are zero
count[isfile]=r_count # except for records that have a matching csv file

# merge csv structs together into a single awkwardarray
pangenome_data=[dat for dat, c in pangenome_data]
pangenome_data=ak.concatenate(pangenome_data)

# unflatten according to csv line counts to fit the main array/df
pangenome_data=ak.unflatten(pangneome_data,count)

# assign pangenome data to array
df['Pangenome']=pangenome_data

Problem is, when I try to save the file in to_parquet
ak.to_parquet(df.Pangenome, datadir+"/up_to_pangenome.parquet", explode_records=True)
I get:
ArrowNotImplementedError: Unhandled type for Arrow to Parquet schema conversion: dense_union<0: string not null=0, 1: string not null=1>

Nonetheless when I try to save the Isolates substructure within the awway:
ak.to_parquet(df.Isolates, datadir+"/up_to_pangenome.parquet", explode_records=True)

I get no error.

When I print the type for ak.Pangenome I get:

ak.type(df.Pangenome):
20 * var * union[{"Gene": option[string],  "Annotation": option[string]}]

But for the isolates I get:

ak.type(df.Isolates)
20 * var * {"IsoNum": float64, "RowNum": int64, "SampleDate": string, "SeqSubDirName": string, "LowQualityThrs": int64, "Freezer_pos": string, "Used": int64, "SiteString": string, "Seqname": string, "IsSameStrain": bool}

Is there a way to convert the union type to a different type within the list to circumvent this?

Thanks!

[jpivarski]

I manually moved this here because the error is a ArrowNotImplementedError, not something I can fix. (You can try the Arrow JIRA.) The discussion would be about ways of working around this (current) Arrow → Parquet limitation.

I'll leave it for others to talk about strategies of turning unions into non-unions, as they'll be prompted by use-cases. Your use-case, however, shouldn't be happening: it's a union of exactly one type:

20 * var * union[{"Gene": option[string],  "Annotation": option[string]}]

shouldn't have been created; it should be

20 * var * {"Gene": option[string],  "Annotation": option[string]}

instead. If you can find the single step that introduces that union, please report it as a bug for me to fix.

Here's how unions are supposed to work. The following must be a union type because it mixes integers and strings:

>>> must_be_union = ak.Array([1, 2, 3, "four", "five"])
>>> must_be_union
<Array [1, 2, 3, 'four', 'five'] type='5 * union[int64, string]'>

The following is still a union type because it could contain multiple types, but this particular value does not.

>>> union_type_but_not_values = must_be_union[:3]
>>> union_type_but_not_values
<Array [1, 2, 3] type='3 * union[int64, string]'>

We may need new functions/tooling to deal with such cases. Similarly, #487 is an open issue to provide tooling for eliminating unwanted option-types.

The strange case you encountered is something I have to build manually because high-level functions are not supposed to produce it:

>>> a = union_type_but_not_values.layout
>>> bad_union = ak.Array(ak.layout.UnionArray8_64(a.tags, a.index, [a.contents[0]]))
>>> bad_union
<Array [1, 2, 3] type='3 * union[int64]'>

If anything like a union[int64] came up in an operation that can return unions, it should have been eliminated by an internal call to simplify:

>>> fixed_it = ak.Array(bad_union.layout.simplify())
>>> fixed_it
<Array [1, 2, 3] type='3 * int64'>

That's why I'd like to find out which of your steps is producing this union of only one type (a record in your example). That one operation is probably not calling simplify as it should.

With your data, though, it should be possible to turn

>>> df_Pangenome = df.Pangenome
>>> ak.type(df_Pangenome):
20 * var * union[{"Gene": option[string],  "Annotation": option[string]}]

into something usable by

>>> fixed = ak.Array(
...     ak.layout.ListArray64(
...         df_Pangenome.layout.starts,
...         df_Pangenome.layout.stops,
...         ak.layout.IndexedArray64(
...             df_Pangenome.layout.content.index,
...             df_Pangenome.layout.content.contents[0],
...         )
...     )
... )
... 
>>> ak.type(fixed)
20 * var * {"Gene": option[string],  "Annotation": option[string]}

but I haven't tested the above (I simulated the Python outputs) because your example is not reproducible (no access to df, for example).

More on UnionArrays here.

[nongiga]

You know what, I am really sorry, I reproduced the example wrong.

Let's add another field called GeneNum.

So the union appears only after concatenation, and it looks like this:

20* union[{"Gene": option[string], "Annotation": option[string], "GeneNum': option[string]}, {"Gene": option[string], "Annotation": option[string], "GeneNum":?int64}]

So it seems that the problem is that pyarrow reads some of the csv's into one type and others into a different type, and that's where the problem originates from. I will try to produce a solution here since it might become a common problem. My guess is that the solution is either loading one csv as an example and setting a schema or merging all the csvs together and only then turning into an ak.Array and unflattering.

Edit:

Alright so problem solved but I encountered another problem with concatenate upon saving. Again I am not sure how to reproduce it but I think it might be a simple fix. Let me know if you want more information and I'll give it to you.

So I have this original code that is the same as above (with the addition that I set the conversion options in advance, in the following manner):

opt=csv.ConvertOptions(column_types=[('Accessory Fragment', pa.string()),( 'Accessory Order with Fragment', pa.string() )],
        strings_can_be_null=True)
 roary_pa=csv.read_csv(filename, opt)

But then when I run it, I get the following error:

----------------------------------------
ArrowInvalidTraceback (most recent call last)
<ipython-input-262-eaff77291036> in <module>
     46 df=ak.with_field(df, Pangenome, where='Pangenome')
     47 
---> 48 ak.to_parquet(df, datadir+"/datastruct/up_to_pangenome.parquet")

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/awkward/operations/convert.py in to_parquet(array, where, explode_records, list_to32, string_to32, bytestring_to32, **options)
   2662     layout = to_layout(array, allow_record=False, allow_other=False)
   2663     iterator = batch_iterator(layout)
-> 2664     first = next(iterator)
   2665 
   2666     if "schema" not in options:

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/awkward/operations/convert.py in batch_iterator(layout)
   2646                         list_to32=list_to32,
   2647                         string_to32=string_to32,
-> 2648                         bytestring_to32=bytestring_to32,
   2649                     )
   2650                 )

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/awkward/operations/convert.py in to_arrow(array, list_to32, string_to32, bytestring_to32)
   2221             )
   2222 
-> 2223     return recurse(layout, None, False)
   2224 
   2225 

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/awkward/operations/convert.py in recurse(layout, mask, is_option)
   1937                     len(offsets) - 1,
   1938                     [None, pyarrow.py_buffer(offsets)],
-> 1939                     children=[content_buffer],
   1940                 )
   1941             else:

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/pyarrow/array.pxi in pyarrow.lib.Array.from_buffers()

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/pyarrow/array.pxi in pyarrow.lib.Array.validate()

~/opt/miniconda3/envs/server-env/lib/python3.7/site-packages/pyarrow/error.pxi in pyarrow.lib.check_status()

ArrowInvalid: Length spanned by list offsets (5305) larger than values array (length 5304)

Since it is an issue with the offsets/values I figured it must be either an issue with concatenate or unflatten. So I mended the code so that all the lines are concatenated by pandas and only then converted to AwkwardArray and unflattened. In this manner I skipped the concatenation step:

datadir=os.getcwd()

# get names of csv files
dirnames=np.char.add(np.asarray(datadir+"/pangenome/", dtype=str), np.asarray(df.CaseNum,dtype=str))
filenames=np.char.add(dirnames,np.asarray('/gene_presence_absence.csv', dtype=str))

# keep track over which of the files exist
isfile=[os.path.isfile(fn) for fn in filenames]

# loop through all existing csv files and get their data into a list of panda structs
roary_pd= [pd.read_csv(fn) for fn in filenames[isfile]]

# get line count in each file for unflattening
count_r= [len(rpd.index) for rpd in roary_pd]
count=np.zeros((len(ilp),),dtype=int) # all line counts are zero
count[isfile]=r_count # except for records that have a matching csv file

# merge csvs together in pandas
merged=pd.concat(roary_pd, ignore_index=True)

#convert from pandas to arrow
par=pa.Table.from_pandas(merged_rpd)

# convert from arrow to AwkwardArray
pangenome_data=ak.from_arrow(par)

# unflatten according to csv line counts to fit the main array/df
pangenome_data=ak.unflatten(pangneome_data,count)

# assign pangenome data to array
df=ak.with_fields(df, pangenome_data, where='Pangenome)
ak.to_parquet(df, datadir+"/up_to_pangenome.parquet")

For the process without ak.concatenate I get the type:

4 * {"NewRandomID": int64, "RandomID": int64, "CaseNum": int64, "Seqplates": option[var * string], "Isolates": var * {"IsoNum": float64, "RowNum": int64, "SampleDate": string, "SeqSubDirName": string, "LowQualityThrs": int64, "Freezer_pos": string, "Used": int64, "SiteString": string, "Seqname": string, "IsSameStrain": bool}, "Pangenome": var * {"Gene": option[string],  "Annotation": option[string]}}

For the process with concatenate I get the type:

4 * {"NewRandomID": int64, "RandomID": int64, "CaseNum": int64, "Seqplates": option[var * string], "Isolates": var * {"IsoNum": float64, "RowNum": int64, "SampleDate": string, "SeqSubDirName": string, "LowQualityThrs": int64, "Freezer_pos": string, "Used": int64, "SiteString": string, "Seqname": string, "IsSameStrain": bool}, "Pangenome": var * {"Gene": option[string],  "Annotation": option[string]}}

So both look identical.

Additionally, converting using to_arrow gives no errors, only the next step does.

[jpivarski]

Knowing that the bad union is being produced by concatenate gives me somewhere to look. I'll try to reproduce it on smaller examples.

As for the second error, Arrow is complaining that the content of a list has unreachable elements, which is allowed in Awkward Array. That's a separate issue (and I know that Arrow doesn't complain about it right away; maybe I should try saving every type to Parquet in the tests). I'll look into that, too.

[gsaha009]

Hi,

Perhaps this issue has already been resolved, but I am kind of facing the same thing while working with awkward array.
Here are the layouts of my two different objects.

# Jet.pt
<ListOffsetArray len='100000'>
    <offsets><Index dtype='int64' len='100001'>
        [     0      2      8 ... 533498 533505 533508]
    </Index></offsets>
    <content><NumpyArray dtype='float32' len='533508'>
        [31.162498 32.81094  31.387499 ... 31.264063 25.385939 17.058594]
    </NumpyArray></content>
</ListOffsetArray>

&

# hcand.pt
<ListArray len='100000'>
    <starts><Index dtype='int64' len='100000'>
        [39588 39588 39588 ... 39588 39588 39588]
    </Index></starts>
    <stops><Index dtype='int64' len='100000'>
        [39588 39588 39588 ... 39588 39588 39588]
    </Index></stops>
    <content><UnionArray len='39588'>
        <tags><Index dtype='int8' len='39588'>[0 0 0 ... 0 0 0]</Index></tags>
        <index><Index dtype='int64' len='39588'>
            [    21  67981     98 ... 676112 524119 676185]
        </Index></index>
        <content index='0'>
            <NumpyArray dtype='float32' len='726294'>
                [22.018867 38.186386 14.039998 ... 27.71415  29.300837
                 10.972406]
            </NumpyArray>
        </content>
        <content index='1'>
            <ListArray len='0'>
                <starts><Index dtype='int64' len='0'>
                    []
                </Index></starts>
                <stops><Index dtype='int64' len='0'>
                    []
                </Index></stops>
                <content><NumpyArray dtype='float32' len='13196'>
                    [34.344795 47.639206 34.660675 ... 44.429535 52.219776
                     50.008633]
                </NumpyArray></content>
            </ListArray>
        </content>
    </UnionArray></content>
</ListArray>

But, I can't write this array in parquet... it is showing the following error

ArrowNotImplementedError                  Traceback (most recent call last)
Cell In[9], line 1
----> 1 ak.to_parquet(events.hcand.pt, "temp.parquet")

File ~/Work/ColumnflowAnalyses/CPinHToTauTau/data/software/venvs/venv_columnar_dev_3cbb5aff/lib/python3.9/site-packages/awkward/_dispatch.py:68, in named_high_level_function.<locals>.dispatch(*args, **kwargs)
     63     else:
     64         raise AssertionError(
     65             "high-level functions should only implement a single yield statement"
     66         )
---> 68 return gen_or_result

File ~/Work/ColumnflowAnalyses/CPinHToTauTau/data/software/venvs/venv_columnar_dev_3cbb5aff/lib/python3.9/site-packages/awkward/_errors.py:67, in ErrorContext.__exit__(self, exception_type, exception_value, traceback)
     60 try:
     61     # Handle caught exception
     62     if (
     63         exception_type is not None
     64         and issubclass(exception_type, Exception)
     65         and self.primary() is self
     66     ):
---> 67         self.handle_exception(exception_type, exception_value)
     68 finally:
     69     # `_kwargs` may hold cyclic references, that we really want to avoid
     70     # as this can lead to large buffers remaining in memory for longer than absolutely necessary
     71     # Let's just clear this, now.
     72     self._kwargs.clear()

File ~/Work/ColumnflowAnalyses/CPinHToTauTau/data/software/venvs/venv_columnar_dev_3cbb5aff/lib/python3.9/site-packages/awkward/_errors.py:82, in ErrorContext.handle_exception(self, cls, exception)
     80     self.decorate_exception(cls, exception)
     81 else:
---> 82     raise self.decorate_exception(cls, exception)

ArrowNotImplementedError: Unhandled type for Arrow to Parquet schema conversion: dense_union<0: extension<awkward<AwkwardArrowType>> not null=0, 1: extension<awkward<AwkwardArrowType>> not null=1>

Please let me know if there is any other way.
Thank you :)

[jpivarski]

This is an Arrow error: there are still some data types that it can't convert to Parquet. In particular, it's unable to write the union of numbers and lists of numbers (like [1, 2, 3, [4, 5]]).

But looking at your example, you probably don't want it to be a union, anyway: you've already prepared the data in such a way that 100% of the elements inside the outer list are numbers; the inner list has length 0. You can write this data, just not with this type.

First, I need to set up something similar to what you have. Your array's data type is "lists of unions of numbers and more lists" (var * union[float32, var * float32]) but in fact the inner union doesn't actually contain any lists. I can make this by first making a union array,

>>> union_array = ak.Array([[1, 2], [3, 4, [5]]])
>>> union_array
<Array [[1, 2], [3, 4, [5]]] type='2 * var * union[int64, var * int64]'>

and then removing the elements that make it a union array:

>>> what_you_have = union_array[:, :2]
>>> what_you_have
<Array [[1, 2], [3, 4]] type='2 * var * union[int64, var * int64]'>

This is like what you have because the type is union, but no values take advantage of the ability to mix 3-levels deep with 2-levels deep. Since no values take advantage of that flexibility, you can cast it as a non-union without error, using ak.enforce_type:

>>> what_you_want = ak.enforce_type(what_you_have, "var * int64")
>>> what_you_want
<Array [[1, 2], [3, 4]] type='2 * var * int64'>

We can also see the difference by looking at the layouts, as you've been doing:

>>> what_you_have.layout
<ListOffsetArray len='2'>
    <offsets><Index dtype='int64' len='3'>
        [0 2 4]
    </Index></offsets>
    <content><UnionArray len='4'>
        <tags><Index dtype='int8' len='4'>
            [0 0 0 0]
        </Index></tags>
        <index><Index dtype='int64' len='4'>
            [0 1 2 3]
        </Index></index>
        <content index='0'>
            <NumpyArray dtype='int64' len='4'>[1 2 3 4]</NumpyArray>
        </content>
        <content index='1'>
            <ListOffsetArray len='1'>
                <offsets><Index dtype='int64' len='2'>[0 1]</Index></offsets>
                <content><NumpyArray dtype='int64' len='1'>[5]</NumpyArray></content>
            </ListOffsetArray>
        </content>
    </UnionArray></content>
</ListOffsetArray>

versus

>>> what_you_want.layout
<ListOffsetArray len='2'>
    <offsets><Index dtype='int64' len='3'>
        [0 2 4]
    </Index></offsets>
    <content><IndexedArray len='4'>
        <index><Index dtype='int64' len='4'>
            [0 1 2 3]
        </Index></index>
        <content><NumpyArray dtype='int64' len='4'>[1 2 3 4]</NumpyArray></content>
    </IndexedArray></content>
</ListOffsetArray>

What you have can't be written to Parquet,

>>> ak.to_parquet(what_you_have, "/tmp/whatever.parquet")
Traceback (most recent call last):
  File "/home/jpivarski/irishep/awkward/src/awkward/_dispatch.py", line 64, in dispatch
    next(gen_or_result)
  File "/home/jpivarski/irishep/awkward/src/awkward/operations/ak_to_parquet.py", line 183, in to_parquet
    return _impl(
  File "/home/jpivarski/irishep/awkward/src/awkward/operations/ak_to_parquet.py", line 416, in _impl
    with pyarrow_parquet.ParquetWriter(
  File "/home/jpivarski/mambaforge/lib/python3.10/site-packages/pyarrow/parquet/core.py", line 1010, in __init__
    self.writer = _parquet.ParquetWriter(
  File "pyarrow/_parquet.pyx", line 2157, in pyarrow._parquet.ParquetWriter.__cinit__
  File "pyarrow/error.pxi", line 154, in pyarrow.lib.pyarrow_internal_check_status
  File "pyarrow/error.pxi", line 91, in pyarrow.lib.check_status
pyarrow.lib.ArrowNotImplementedError: Unhandled type for Arrow to Parquet schema conversion: dense_union<0: extension<awkward<AwkwardArrowType>> not null=0, 1: extension<awkward<AwkwardArrowType>> not null=1>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/jpivarski/irishep/awkward/src/awkward/_dispatch.py", line 38, in dispatch
    with OperationErrorContext(name, args, kwargs):
  File "/home/jpivarski/irishep/awkward/src/awkward/_errors.py", line 85, in __exit__
    self.handle_exception(exception_type, exception_value)
  File "/home/jpivarski/irishep/awkward/src/awkward/_errors.py", line 95, in handle_exception
    raise self.decorate_exception(cls, exception)
pyarrow.lib.ArrowNotImplementedError: Unhandled type for Arrow to Parquet schema conversion: dense_union<0: extension<awkward<AwkwardArrowType>> not null=0, 1: extension<awkward<AwkwardArrowType>> not null=1>

See if this has been reported at https://github.com/scikit-hep/awkward/issues

but what you want can be:

>>> ak.to_parquet(what_you_want, "/tmp/whatever.parquet")
<pyarrow._parquet.FileMetaData object at 0x78b9196b2bb0>
  created_by: parquet-cpp-arrow version 15.0.0
  num_columns: 1
  num_rows: 2
  num_row_groups: 1
  format_version: 2.6
  serialized_size: 0

[gsaha009]

Hi @jpivarski
Thanks a lot for this prompt solution with this explanation. This enforce_type is working really smooth.
Gourab