Variable OCCURS clause

[tr11]

Describe the bug

Variable OCCURS fails if we don't specify variable record lengths. This may be very similar to the discussion of #156, but I think the issue there is slightly different.

To Reproduce

Run the following with pyspark

import os
import tempfile
from pyspark.sql.functions import explode

with tempfile.NamedTemporaryFile('wb') as f:
    f.write(b'   5ABC1ABC2ABC3ABC4ABC5   5DEF1DEF2DEF3DEF4DEF5')
    f.flush()

    (spark
      .read
      .format("cobol")
      .option("copybook_contents", """
          01 RECORD.
              02 COUNT PIC 9(4).
              02 GROUP OCCURS 0 TO 5 TIMES DEPENDING ON COUNT.
                  03 TEXT   PIC X(3).
                  03 FIELD  PIC 9.
      """)
      .option("encoding", 'ascii')
      .option("variable_size_occurs", "true")
      .load(f.name)
    ).select('RECORD.COUNT', explode('RECORD.GROUP')).show()

    (spark
      .read
      .format("cobol")
      .option("copybook_contents", """
          01 RECORD.
              02 COUNT PIC 9(4).
              02 GROUP OCCURS 0 TO 11 TIMES DEPENDING ON COUNT.
                  03 TEXT   PIC X(3).
                  03 FIELD  PIC 9.
      """)
      .option("encoding", 'ascii')
      .option("variable_size_occurs", "true")
      .load(f.name)
    ).select('RECORD.COUNT', explode('RECORD.GROUP')).show()    
    
    (spark
      .read
      .format("cobol")
      .option("copybook_contents", """
          01 RECORD.
              02 COUNT PIC 9(4).
              02 GROUP OCCURS 0 TO 10 TIMES DEPENDING ON COUNT.
                  03 TEXT   PIC X(3).
                  03 FIELD  PIC 9.
      """)
      .option("encoding", 'ascii')
      .option("variable_size_occurs", "true")
      .load(f.name)
    ).select('RECORD.COUNT', explode('RECORD.GROUP')).show()    

Expected behavior

All examples above should return the same

+------------+--------+
|RECORD.COUNT|     col|
+------------+--------+
|           5|[ABC, 1]|
|           5|[ABC, 2]|
|           5|[ABC, 3]|
|           5|[ABC, 4]|
|           5|[ABC, 5]|
|           5|[DEF, 1]|
|           5|[DEF, 2]|
|           5|[DEF, 3]|
|           5|[DEF, 4]|
|           5|[DEF, 5]|
+------------+--------+

but the second one gives only the first record

+------------+--------+
|RECORD.COUNT|     col|
+------------+--------+
|           5|[ABC, 1]|
|           5|[ABC, 2]|
|           5|[ABC, 3]|
|           5|[ABC, 4]|
|           5|[ABC, 5]|
+------------+--------+

and the third one fails with a file size check.

Additional context

This issue seems to be a consequence of the fact that the files I have to process do not have a leading RDW block or a field specifying record lengths. Essentially, the variable OCCURS clause means that in my case the record needs to be read in full, but then we need to backtrack the number of bytes that were not needed before reading the following record.

[yruslan]

Variable size OCCURS produces variable size records. And variable size records are only supported if records have an RDW header or a record length field.

The idea of backtracking is interesting and might be the way to overcome the limitation. At first glance, it seems doable. I will think about the details next week.

The issue to solve is related to scale processing of such files. Currently a sparse index is generated based on RDW. That allows to have a way to split a variable record length file between partitions. But the index generation happens before records are parsed so actual array sizes are unknown at that stage.

To implement the backtracking the reader needs to

• Gather all information on the locations of array element sizes and array lengths and the ordering and nesting of arrays,
• This info needs to be used when a sparse index is generated so that files are split properly between partitions.
• During the record decoding stage the backtracking algorithm needs to be implemented.

[tr11]

That's exactly what I was thinking. I did something similar for a custom parser I built in the past for files without RDW headers. What it does is the following (recursively): 1. If the group or subgroups contain no OCCURS, the size is known from the copybook directly (i.e. group of fixed length) 2. Otherwise, determine the size of subgroups and gather all the depended-on fields. The group size is the sum of the products of those values. It's not as efficient as reading the RDW, but the result is the record size that can be used to build the index. I can implement this is you think this is the way to go. Backtracking would be hard to implement with partitions. It works fine serially but unless we know the exact start of the first record in the partition we are kind of stuck.

…

On Sat, Jan 18, 2020, 06:39 Ruslan Yushchenko ***@***.***> wrote: Variable size OCCURS produces variable size records. And variable size records are only supported if records have an RDW header or a record length field. The idea of backtracking is interesting and might be the way to overcome the limitation. At first glance, it seems doable. I will think about the details next week. The issue to solve is related to scale processing of such files. Currently a sparse index is generated based on RDW. That allows to have a way to split a variable record length file between partitions. But the index generation happens before records are parsed so actual array sizes are unknown at that stage. To implement the backtracking the reader needs to - Gather all information on the locations of array element sizes and array lengths and the ordering and nesting of arrays, - This info needs to be used when a sparse index is generated so that files are split properly between partitions. - During the record decoding stage the backtracking algorithm needs to be implemented. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#239?email_source=notifications&email_token=AAJ6T2K57G37CXIBMT37SI3Q6LS7JA5CNFSM4KIJUV3KYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEJJWNQI#issuecomment-575891137>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ6T2MAJ3PMHAWOH5OZHC3Q6LS7JANCNFSM4KIJUV3A> .

[tr11]

In fact, something like this could even simplify the code as there is no need for a distinction between fixed- and variable-length records; length would be determined by:

1. existence of an RDW block (using the same header parser we use today)
2. a field within the record
3. figured out from the record itself by parsing the OCCURS dependents

[yruslan]

Yes, it seems like a very good idea to have a special custom record header parsers for a situation when variable_size_occurs=true and no RDWs, and no record length field.

• That parser could fetch the input stream up to a first occurs size field, calculate the next chunk, fetch data up to the next occurs size field, and so forth.
• And finally, it could fetch the rest of the data up to the end of the record.
• Since record header parsers are used by the index builder automatically, no other logic needs to be updated.

In fact, something like this could even simplify the code as there is no need for a distinction between fixed- and variable-length records; length would be determined by:

1. existence of an RDW block (using the same header parser we use today)
2. a field within the record [size]
3. figured out from the record itself by parsing the OCCURS dependents

The 1-2 is how it already works. Yes, we may not distinguish between fixed record length files and variable record length files. But the simplest fixed record length files use Spark's binaryRecords() API and are still more efficient than any of the instances where record sizes are variable.

[tr11]

Maybe we could change the logic of computing BinaryProperties to also keep track of offsets of dependedOn fields. What I meant was that we wouldn't have to tell cobrix explicitly whether the record is fixed or not. If there is no rdw, no length record, and no variable occurs, then we have fixed length records, otherwise variable. I guess I'm proposing eliminating the variable records option and let the parser decide based on the above.

…

On Mon, Jan 20, 2020, 02:49 Ruslan Yushchenko ***@***.***> wrote: Yes, it seems like a very good idea to have a special custom record header parsers for a situation when variable_size_occurs=true and no RDWs, and no record length field. - That parser could fetch the input stream up to a first occurs size field, calculate the next chunk, fetch data up to the next occurs size field, and so forth. - And finally, it could fetch the rest of the data up to the end of the record. - Since record header parsers are used by the index builder automatically, no other logic needs to be updated. In fact, something like this could even simplify the code as there is no need for a distinction between fixed- and variable-length records; length would be determined by: 1. existence of an RDW block (using the same header parser we use today) 2. a field within the record [size] 3. figured out from the record itself by parsing the OCCURS dependents The 1-2 is how it already works. Yes, we may not distinguish between fixed record length files and variable record length files. But the simplest fixed record length files use Spark's binaryRecords() API and are still more efficient than any of the instances where record sizes are variable. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#239?email_source=notifications&email_token=AAJ6T2LXBRZUP3I32MLCXZDQ6VJSJA5CNFSM4KIJUV3KYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEJLVJSA#issuecomment-576148680>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAJ6T2IX65BS5LGJXQXM35DQ6VJSJANCNFSM4KIJUV3A> .

[yruslan]

Maybe we could change the logic of computing BinaryProperties to also keep
track of offsets of dependedOn fields.

Yes, might be helpful.
I'll probably go with an underlying custom record parser implementation. The parser will be used in case a file without RDWs or a record length field. The parser will be provided with offsets to dependent fields and types and corresponding OCCURS sizes.

I'll look into this issue in a week or two. There are some high priority things I need to finish.

What I meant was that we wouldn't have to tell cobrix explicitly whether
the record is fixed or not.

Yes, this is exactly how it works now. The .option("is_record_sequence", "true") specifies that the file has RDWs. The name is misleading. Maybe we should add a synonym, say, has_rdw (or similar). Funnily enough, this option already has a synonym is_xcom since the headers added to the beginning of each record were thought to be added by XCOM. 😄

[yruslan]

• Is it possible to have several variable-length arrays in one record in your use case?
• Is it possible to have a variable-length array inside a variable-length array in your use case?

Trying to figure out if support for simple cases is sufficient or we need a more generic implementasion.

[tr11]

Yeah, I have a few files that have multiple variable occurs, with some of those nested!

What I implemented in the past that I can do here too is a recursive size method to give the size of group given a record. The output would be something akin to Option[int, struct of offsets and datatypes] that could be applied to a record. If there are no variable occurs then it's just an int and we can skip the record.

[yruslan]

The way the custom record header interface is implemented it has the following limitations:

• The number of bytes to determine the size of the record should be known in advance and it cannot be bigger than the size of the current record. For multiple nested variable-size arrays this size cannot generally be determined.
• The buffer used is a forward-only buffer. If a record is read past its end the current position in the buffer cannot be moved back.
  Give the above limitations a new interface for custom record header parsers need to be implemented in order to support this use case.

I'm wondering why such complicated files do not contain RDWs or a record length field?
Is it a limitation of a tool that transfers files from mainframes?

[yruslan]

In order to implement this use case scalably and as generic as it is, we need a sparse index builder that takes an AST and an instance of SimpleStream and returns a sparse index in form of ArrayBuffer[SparseIndexEntry].
Basically IndexGenerator.sparseIndexGenerator should be rewritten to account for this particular case.

A very helpful thing from you side would be a unit test that uses a copybook with several variable-size arrays, and a data file with various array sizes, including 0, so that the feature can be easily tested once implemented.

[tr11]

I'm wondering why such complicated files do not contain RDWs or a record length field?
Is it a limitation of a tool that transfers files from mainframes?

These are files that we receive from external vendors and have no control over. They structure the files such that there's a 10-byte header that defines the segment. From there we can determine the max size of the record, but we will only know as we read it.

[tr11]

In order to implement this use case scalably and as generic as it is, we need a sparse index builder that takes an AST and an instance of SimpleStream and returns a sparse index in form of ArrayBuffer[SparseIndexEntry].
Basically IndexGenerator.sparseIndexGenerator should be rewritten to account for this particular case.

To make sure, we're doing