Issue 1045 - Allow downcasting of parquet timestamp[ns] to timestamp[us] for Table.add_files()

[lloyd-EA]

When calling Table.add_files(), the env PYICEBERG_DOWNCAST_NS_TIMESTAMP_TO_US_ON_WRITE isn't being used on _check_pyarrow_schema_compatible, therefore, it fails for parquet files with ns timestamps.

Credits to @fussion2222 for suggesting this fix.

[kevinjqliu]

Thanks for the PR!

Generally LGTM, another option is to pass the option down from add_files.

Could you also add a test case for this?
Something similar to,

iceberg-python/tests/integration/test_add_files.py

Line 588 in 3b53edc

def test_add_files_with_timestamp_tz_ns_fails(session_catalog: Catalog, format_version: int, mocker: MockerFixture) -> None:

[lloyd-EA]

@kevinjqliu Thanks for having a look.

I've updated the PR and added a test case for it as requested.

I'm open for suggestions and corrections. Thanks!

[sungwy]

Thanks for working on this PR @lloyd-EA and thank you @kevinjqliu for the review. Could I add another suggestion to read the table using Spark and verify that the value that's been read is equal to the expected value?

As I mentioned in a comment in #1045 (comment) I think it would be great to verify that the data is readable as us timestamp through a Spark session as well. Here's an example that uses a spark session for reference:

iceberg-python/tests/integration/test_add_files.py

Lines 154 to 177 in 9850290

	def test_add_files_to_unpartitioned_table(spark: SparkSession, session_catalog: Catalog, format_version: int) -> None:
	identifier = f"default.unpartitioned_table_v{format_version}"
	tbl = _create_table(session_catalog, identifier, format_version)
	file_paths = [f"s3://warehouse/default/unpartitioned/v{format_version}/test-{i}.parquet" for i in range(5)]
	# write parquet files
	for file_path in file_paths:
	fo = tbl.io.new_output(file_path)
	with fo.create(overwrite=True) as fos:
	with pq.ParquetWriter(fos, schema=ARROW_SCHEMA) as writer:
	writer.write_table(ARROW_TABLE)
	# add the parquet files as data files
	tbl.add_files(file_paths=file_paths)
	# NameMapping must have been set to enable reads
	assert tbl.name_mapping() is not None
	rows = spark.sql(
	f"""
	SELECT added_data_files_count, existing_data_files_count, deleted_data_files_count
	FROM {identifier}.all_manifests
	"""
	).collect()

[lloyd-EA]

@sungwy I updated the PR there. It seems that Spark reads in the column as us and doesn't check the underlying data file's precision being ns, so when it is converted to a datetime the value is wrong.

expected: 2021-03-17 07:54:47.249846175

To me, this is a problem with Spark not being able to handle ns timestamps properly, and since I won't be using Spark for my application, I'll continue to use my forked repo for now as it's sufficient for my application.

[sungwy]

Thank you for following up on the suggestion @lloyd-EA . I'm glad we included this integration test to find out the issue with Spark's timestamp conversion.

I think we will have to investigate the corresponding schema evolution handling logic in Spark-Iceberg to understand if this is a bug, or an expected behavior with Spark.

If it is not possible to fix this issue in Spark-Iceberg, I think it'll mean that we will not be able to support adding ns timestamp files until V3 spec is supported.

Please do keep in mind that Iceberg spec is a language agnostic spec, and hence ruling out the possibility of using the same Iceberg table with other tools (like Trino or Spark) may impact you down the line when your application system grows.