diff --git a/daft/dataframe/dataframe.py b/daft/dataframe/dataframe.py index 06bf0dd2a0..9a73db9e00 100644 --- a/daft/dataframe/dataframe.py +++ b/daft/dataframe/dataframe.py @@ -948,6 +948,18 @@ def write_deltalake( from daft.io._deltalake import large_dtypes_kwargs from daft.io.object_store_options import io_config_to_storage_options + # to ensure compatibility for deltalake<0.20 + try: + from deltalake import CommitProperties + + def _create_custom_metadata(metadata: Optional[Dict[str, str]]): + return CommitProperties(custom_metadata=custom_metadata) + except ImportError: + CommitProperties = None + + def _create_custom_metadata(metadata: Optional[Dict[str, str]]): + return metadata + if schema_mode == "merge": raise ValueError("Schema mode' merge' is not currently supported for write_deltalake.") @@ -1093,7 +1105,12 @@ def write_deltalake( sizes.append(old_actions_dict["size_bytes"][i]) table._table.create_write_transaction( - add_actions, mode, partition_cols or [], delta_schema, None, custom_metadata + add_actions, + mode, + partition_cols or [], + delta_schema, + None, + _create_custom_metadata(custom_metadata), ) table.update_incremental()