read_hudi fails on next(iter(latest_commit_metadata["partitionToWriteStats"].items()))

[chatnord]

Hi all,
I am trying to read a Hudi table from S3 and I get the following error:

StopIteration                             Traceback (most recent call last)
Cell In[28], line 5
      2 import daft
      4 io_config = daft.io.IOConfig(s3=daft.io.S3Config(key_id=KEY, access_key=SECRET))
----> 5 df = daft.read_hudi(path, io_config=io_config)
      6 df.show()

    [... skipping hidden 2 frame]

File ~/.pyenv/versions/3.10.0/lib/python3.10/site-packages/daft/io/_hudi.py:38, in read_hudi(table_uri, io_config)
     35 multithreaded_io = not context.get_context().is_ray_runner
     36 storage_config = StorageConfig.native(NativeStorageConfig(multithreaded_io, io_config))
---> 38 hudi_operator = HudiScanOperator(table_uri, storage_config=storage_config)
     40 handle = ScanOperatorHandle.from_python_scan_operator(hudi_operator)
     41 builder = LogicalPlanBuilder.from_tabular_scan(scan_operator=handle)

File ~/.pyenv/versions/3.10.0/lib/python3.10/site-packages/daft/hudi/hudi_scan.py:29, in HudiScanOperator.__init__(self, table_uri, storage_config)
     27 self._table = HudiTable(table_uri, resolved_fs, resolved_path[0])
     28 self._storage_config = storage_config
---> 29 self._schema = Schema.from_pyarrow_schema(self._table.schema)
     30 partition_fields = set(self._table.props.partition_fields)
     31 self._partition_keys = [
     32     PartitionField(field._field) for field in self._schema if field.name in partition_fields
     33 ]

File ~/.pyenv/versions/3.10.0/lib/python3.10/site-packages/daft/hudi/pyhudi/table.py:195, in HudiTable.schema(self)
    193 @property
    194 def schema(self) -> pa.Schema:
--> 195     return self._meta_client.get_active_timeline().get_latest_commit_schema()

File ~/.pyenv/versions/3.10.0/lib/python3.10/site-packages/daft/hudi/pyhudi/timeline.py:63, in Timeline.get_latest_commit_schema(self)
     61 def get_latest_commit_schema(self) -> pa.Schema:
     62     latest_commit_metadata = self.get_latest_commit_metadata()
---> 63     _, write_stats = next(iter(latest_commit_metadata["partitionToWriteStats"].items()))
     64     base_file_path = os.path.join(self.base_path, write_stats[0]["path"])
     65     with self.fs.open_input_file(base_file_path) as f:

StopIteration: 

The error message does not really help debugging, so I am a bit lost. The original table does not have stats, I wonder if this must be it?

[jaychia]

Hi @chatnord!

It seems that next(iter(latest_commit_metadata["partitionToWriteStats"].items())) is throwing an error, which is probably because latest_commit_metadata["partitionToWriteStats"] is an empty dictionary.

I'm not as familiar with Hudi internals, let me cc @xushiyan on the Hudi side of things for his thoughts

[xushiyan]

@chatnord do you mind zipping and sharing your hudi table's .hoodie/ folder? it's only metadata about the table which should not include any sensitive info

[xushiyan]

@chatnord in normal cases, partitionToWriteStats should be present in writer commits. I'm curious to see your timeline state. Nevertheless, this PR should address the issue.

[chatnord]

Hey, sorry but I was away. Not sure I can share the metadata, as the tables are confidential. I can mock it, but it will take a bit. However, this table does not have stats - possibly this is the issue, and I agree that this PR seems to be the solution.

[xushiyan]

@chatnord are we good to close this?

[chatnord]

Sure, thank you for the support!