(schema, batch_gen)
| 656 | |
| 657 | |
| 658 | def make_udt_func(schema, batch_gen): |
| 659 | def udf_func(ctx): |
| 660 | class UDT: |
| 661 | def __init__(self): |
| 662 | self.caller = None |
| 663 | |
| 664 | def __call__(self, ctx): |
| 665 | try: |
| 666 | if self.caller is None: |
| 667 | self.caller, ctx = batch_gen(ctx).send, None |
| 668 | batch = self.caller(ctx) |
| 669 | except StopIteration: |
| 670 | arrays = [pa.array([], type=field.type) |
| 671 | for field in schema] |
| 672 | batch = pa.RecordBatch.from_arrays( |
| 673 | arrays=arrays, schema=schema) |
| 674 | return batch.to_struct_array() |
| 675 | return UDT() |
| 676 | return udf_func |
| 677 | |
| 678 | |
| 679 | def datasource1_direct(): |
no outgoing calls
no test coverage detected