bug: tolerating metadata written by different engines with Delta Lake

[djouallah]

i was appending data to a delta table, worked fine before, when i upgraded, I am getting the following error, notice, it is saving to an abfss location

QueryExecutionException                   Traceback (most recent call last)
File <timed exec>:16

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/readwriter.py:679, in DataFrameWriter.save(self, path, format, mode, partitionBy, **options)
    677     self.format(format)
    678 self._write.path = path
--> 679 _, _, ei = self._spark.client.execute_command(
    680     self._write.command(self._spark.client), self._write.observations
    681 )
    682 self._callback(ei)

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/client/core.py:1148, in SparkConnectClient.execute_command(self, command, observations)
   1146     req.user_context.user_id = self._user_id
   1147 req.plan.command.CopyFrom(command)
-> 1148 data, _, metrics, observed_metrics, properties = self._execute_and_fetch(
   1149     req, observations or {}
   1150 )
   1151 # Create a query execution object.
   1152 ei = ExecutionInfo(metrics, observed_metrics)

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/client/core.py:1560, in SparkConnectClient._execute_and_fetch(self, req, observations, self_destruct)
   1557 properties: Dict[str, Any] = {}
   1559 with Progress(handlers=self._progress_handlers, operation_id=req.operation_id) as progress:
-> 1560     for response in self._execute_and_fetch_as_iterator(
   1561         req, observations, progress=progress
   1562     ):
   1563         if isinstance(response, StructType):
   1564             schema = response

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/client/core.py:1537, in SparkConnectClient._execute_and_fetch_as_iterator(self, req, observations, progress)
   1535     raise kb
   1536 except Exception as error:
-> 1537     self._handle_error(error)

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/client/core.py:1811, in SparkConnectClient._handle_error(self, error)
   1809     self.thread_local.inside_error_handling = True
   1810     if isinstance(error, grpc.RpcError):
-> 1811         self._handle_rpc_error(error)
   1812     raise error
   1813 finally:

File ~/jupyter-env/python3.11/lib/python3.11/site-packages/pyspark/sql/connect/client/core.py:1882, in SparkConnectClient._handle_rpc_error(self, rpc_error)
   1879             if info.metadata["errorClass"] == "INVALID_HANDLE.SESSION_CHANGED":
   1880                 self._closed = True
-> 1882             raise convert_exception(
   1883                 info,
   1884                 status.message,
   1885                 self._fetch_enriched_error(info),
   1886                 self._display_server_stack_trace(),
   1887             ) from None
   1889     raise SparkConnectGrpcException(status.message) from None
   1890 else:

QueryExecutionException: Delta transaction failed, version 0 already exists.

[shehabgamin]

@djouallah Thanks for reporting, we'll look into this now and do another release after fixing

[lonless9]

Hi @djouallah Thanks for sharing the details.
Based on your description, we have a guess about your scenario. Could you confirm if this is what you were doing?

For an existing table:

df2.write.format("delta").save(table_path) # This would fail with that error

df2.write.format("delta").mode("append").save(table_path) # This would work

This is because when the save mode is not specified, we use ErrorIfExists by default.

[djouallah]

I am using this spark.createDataFrame(df).write.format("delta").mode("append").save(results)

[linhr]

Could you share a complete example? We were not able to reproduce this yet.

[djouallah]

sure https://github.com/djouallah/Fabric_Notebooks_Demo/blob/main/sparkconnect/TPCH_lakesail.ipynb

[linhr]

Thanks for sharing the notebook!

Is this location written already before the upgrade? Or is it a new table location for appending data?

If you change results to a different path (which is never written before) and append data multiple times, would there still be an error?

(I'm trying to get clues since we haven't been able to reproduce it locally yet. I tried the abfss protocol using the Azurite emulator.)

[djouallah]

it an existing table location, and it is written by different writers , it a central location to store results , using a new location and appending multiple times works fine

[shehabgamin]

@djouallah We are still unable to reproduce. If you switch back to Sail 0.3.3 does the error go away?

We are releasing Sail 0.3.5 in the next few hours which includes #837, but I have no reason to believe that the fixes in the linked PR are relevant to the issue you are encountering.

[linhr]

If the data in the existing location is still available, would you mind sharing it with us (if there is no sensitive information) so that we can better diagnose the particular issue? You can upload it here, ping us in our community Slack, or reach us via email (hello@lakesail.com).

I also tried the following with our latest code and did not see the issue either.

• Appending to a table written by deltalake.
• Appending to a table written by the earlier version pysail==0.3.3.

Do you remember which writer you were using when writing to the location originally?

[djouallah]

Not a problem at all, I tried to reproduce it using a new location and it works fine

[linhr]

Thanks for sharing the data via email! I can reproduce the behavior now. We'll take a look.

[linhr]

It turns out that there are checkpoint files with non-canonical schema in this particular dataset. Our recent changes (both our own changes and upgrades of the Delta library) may have introduced stricter validation of the existing table, causing the append operation to fail.

Specifically, the following code would treat the table as "non-existent" when loading existing table fails. This would cause the writer to write version 0, which would fail since version 0 actually exists.

sail/crates/sail-data-source/src/formats/delta.rs

Lines 74 to 77 in 7b1d1ad

	let table_exists =
	open_table_with_object_store(table_url.clone(), object_store, Default::default())
	.await
	.is_ok();

The internal error when loading table is the following:

Failed to convert into Arrow schema: Invalid argument error: It is not possible to concatenate arrays of different data types (Struct(partitionValues Map(Field { name: "key_value", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false)), Struct(partitionValues Map(Field { name: "entries", data_type: Struct([Field { name: "key", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: "value", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }]), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, false))).

It turns out that the Delta Rust library would load Arrow map data with the key_value field name when reading _delta_log/0000000000000000XXXX.json files. But there are also checkpoint files _delta_log/0000000000000000XXXX.checkpoint.parquet where Arrow map data is stored with the entries field name. Merging these incompatible metadata files results in an error when building the snapshot.

There are a few improvements we could do:

1. Improve error reporting for Delta table loading. The code I linked above does not differentiate among different types of errors when trying to load an existing table. If the error is other than "not found", we should fail fast.
2. Be more tolerant on the metadata written by different engines. The field name for Arrow map/list types should usually be ignored when working with Arrow.

[djouallah]

@linhr thanks, appreciate your investigations, trying to write to the same table using multiple writer in delta is an act of faith :)