Cast 's', 'ms' and 'ns' PyArrow timestamp to 'us' precision on write

[sungwy]

Closes #541 and #840

Question: Are timestamp_ns and timestamptz_ns already supported? If so, should we just limit this PR to casting 's' and 'ms' to 'us' precision, and instead introduce the new timestamp_ns types?

[Fokko]

Thanks @syun64 for working on this! 🙌

Question: Are timestamp_ns and timestamptz_ns already supported? If so, should we just limit this PR to casting 's' and 'ms' to 'us' precision, and instead introduce the new timestamp_ns types?

Nanoseconds timestamp is supported in V3. In order to write nanoseconds without downcasting, we need to check if it is a V3 table.

[HonahX]

@syun64 It is great to have an optional flag to add more compatibility around nanosecond timestamp before V3. Thanks for working on this! I have one comment on the effect of this change on read side. Please let me know what you think!

[HonahX]

How about making downcast_ns_timestamp a parameter of schema_to_pyarrow(TYPO: should be pyarrow_to_schema), and reading the Config from yml when we use this API on write? schema_to_pyarrow(TYPO: should be pyarrow_to_schema) itself seems to be a useful public API so it may be good to explicitly reveal the optional downcast. This will also help mitigate an edge case:

Since pyarrow_to_schema is used for both read/write, enabling this option also allows unit ns to pass the schema conversion when reading. For example, If users add a parquet file with ns timestamp and try to read the table as arrow, they will find the read process pass the pyarrow_to_schema check and stops at to_request_schema with

 pyarrow.lib.ArrowInvalid: Casting from timestamp[ns] to timestamp[us] would lose data:

[sungwy]

Thank you for raising this @HonahX - I think this is an important bad case to consider

I actually don't think it'll stop when it reads through to_requested_schema, because it will detect that the pyarrow types are different, but their IcebergTypes are the same and silently cast on read, which will drop the precision silently:

iceberg-python/pyiceberg/io/pyarrow.py

Line 1274 in 0e381fa

elif (target_type := schema_to_pyarrow(field.field_type, include_field_ids=False)) != values.type:

This logic was introduced to support casting small and large types interchangeably: different pyarrow types that can be mapped to the same IcebergType (string, large_string) can be cast and read through as the same PyArrow type

The only thing that blocks this write from succeeding currently is the pyarrow_to_schema call which fails to generate a corresponding IcebergSchema based on the provided pyarrow schema, which this PR seeks to fix.

I do think that the silent downcasting of data is problematic - but that isn't the only problematic aspect of the add_files API. add_files does not check for the validity of the schema, because we provide a list of files into the API. Currently, it is up to the user to ensure that the file they want to add is in the correct format, and own the risk of potentially introducing wrong data files into the table. We note that the API is only intended for expert users, which is similar to the warnings we have for the other existing Table migration procedures.

Do you think it would be helpful to decouple this concern to that of the idea of introducing an optional schema check for the add_files procedure?

[sungwy]

On a tangent, I'd like to raise another point for discussion:

If we are aware that nanoseconds will be introduced as a separate IcebergType, would introducing a pa.timestamp(unit="ns") -> TimestampType introduce too much complexity, since we will have to maintain the logic for one to many mapping for pa.timestamp(unit="ns") -> TimestampType, TimestampNsType based on the format-version of the Iceberg table? Is introducing automated conversion of ns precision timestamp really worth the complexity we are introducing in the near future?

[corleyma]

I still think @HonahX raises a good point about the schema_to_pyarrow method being a useful public API, and it would be nice for its behavior to not be too tightly coupled to pyiceberg config. I.e., I agree that it's wiser to parameterize the behavior and determine the correct parameter to use via config where it's called.

[sungwy]

Thank you for your input, @corleyma. Just clarifying here - what enables us to write ns into TimestampType in PyIceberg is this proposed change in ConvertToIceberg, which is not in schema_to_pyarrow. It is actually in pyarrow_to_schema which is used to check schema compatibility on write. Once the data file is written, we are making the assumption that TimestampType is all in 'us' precision, or that it is safe to cast to 'us' precision, because the writer has already made the decision to write into 'us' precision timestamps.

If we are aware that nanoseconds will be introduced as a separate IcebergType, would introducing a pa.timestamp(unit="ns") -> TimestampType introduce too much complexity, since we will have to maintain the logic for one to many mapping for pa.timestamp(unit="ns") -> TimestampType, TimestampNsType based on the format-version of the Iceberg table? Is introducing automated conversion of ns precision timestamp really worth the complexity we are introducing in the near future?

@Fokko , @HonahX and @corleyma : I'd like to gather some feedback on this point before committing to introducing this flag. My worry is that since there's a new type that's being introduced in V3 Spec that will actually be in 'ns', enabling 'ns' casting on the existing 'us' precision TimestampType will complicate the type conversions, dooming us to have to check the type (TimestampType, TimestampNsType), downcast-to-ns boolean flag, and the format-version whenever we are casting timestamps. I'd like for us to weigh that trade off carefully and decide on whether supporting this conversion is worth the complexity we are introducing into the conversion functions.

[Fokko]

Thanks @HonahX for giving the example, I just gave this a spin and ran into the following:

@pytest.mark.integration
def test_timestamp_tz(
    session_catalog: Catalog, format_version: int, mocker: MockerFixture
) -> None:
    nanoseconds_schema_iceberg = Schema(
        NestedField(1, "quux", TimestamptzType())
    )

    nanoseconds_schema = pa.schema([
        ("quux", pa.timestamp("ns", tz="UTC")),
    ])

    arrow_table = pa.Table.from_pylist(
        [
            {
                "quux": 1615967687249846175,  # 2021-03-17 07:54:47.249846159
            }
        ],
        schema=nanoseconds_schema,
    )
    mocker.patch.dict(os.environ, values={"PYICEBERG_DOWNCAST_NS_TIMESTAMP_ON_WRITE": "True"})

    identifier = f"default.abccccc{format_version}"

    try:
        session_catalog.drop_table(identifier=identifier)
    except NoSuchTableError:
        pass

    tbl = session_catalog.create_table(
        identifier=identifier,
        schema=nanoseconds_schema_iceberg,
        properties={"format-version": str(format_version)},
        partition_spec=PartitionSpec(),
    )

    file_paths = [f"s3://warehouse/default/test_timestamp_tz/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=nanoseconds_schema) as writer:
                writer.write_table(arrow_table)

    # add the parquet files as data files
    tbl.add_files(file_paths=file_paths)

    print(tbl.scan().to_arrow())

I think we can force the cast to be unsafe:

return values.cast(target_type, safe=False)

We might want to check if we only apply this when doing the nanos to micros. I'm not sure what will happen when we do other lossy conversions.

[Fokko]

I also got some issues with the nanosecond timestamp when collecting statistics:

>   ???
E   ValueError: Nanosecond resolution temporal type 1615967687249846175 is not safely convertible to microseconds to convert to datetime.datetime. Install pandas to return as Timestamp with nanosecond support or access the .value attribute.

At the lines:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1870 to 1871 in 7afd6d6

	col_aggs[field_id].update_min(statistics.min)
	col_aggs[field_id].update_max(statistics.max)

This got fixed after updating this to:

                    col_aggs[field_id].update_min(statistics.min_raw)
                    col_aggs[field_id].update_max(statistics.max_raw)

[sungwy]

Hi folks - thank you all for the valuable feedback. So it sounds like what we want is for the flag to be controlled by the configuration flag, but that flag to be passed as a parameter to the schema_to_pyarrow API so that its behavior can be fully controlled by its input parameters.

I've made the following changes:

1. Introduced downcast_ns_timestamp_to_us as a new input parameter to pyarrow_to_schema and to_requested_schema public APIs
2. Now table and catalog level functions infer the flag from the Config on write. (e.g. _check_schema_compatible and _convert_schema_if_needed)
3. Always downcast ns to us on read, if there is ns timestamp in the parquet file (we will want to revise this behavior when we introduce nanosecond support in V3 spec, but until then, I think it's a reasonable assumption that data files that are in Iceberg will only be read with microseconds precision). https://github.com/apache/iceberg-python/pull/848/files#diff-8d5e63f2a87ead8cebe2fd8ac5dcf2198d229f01e16bb9e06e21f7277c328abdR1030-R1033

[sungwy]

I also got some issues with the nanosecond timestamp when collecting statistics:

>   ???
E   ValueError: Nanosecond resolution temporal type 1615967687249846175 is not safely convertible to microseconds to convert to datetime.datetime. Install pandas to return as Timestamp with nanosecond support or access the .value attribute.

At the lines:

iceberg-python/pyiceberg/io/pyarrow.py

Lines 1870 to 1871 in 7afd6d6

	col_aggs[field_id].update_min(statistics.min)
	col_aggs[field_id].update_max(statistics.max)

This got fixed after updating this to:

                    col_aggs[field_id].update_min(statistics.min_raw)
                    col_aggs[field_id].update_max(statistics.max_raw)

I tried making this change and realized that this causes our serialization to break because it introduces bytes values in our statistics, which cannot be serialized (since it already is). I will need to spend a bit more time to figure out the right change to StatsAggregator to support this change. I also failed to reproduce this issue in my environment (possibly because it has pandas installed) so I'm reverting this change for now.

[Fokko]

I also failed to reproduce this issue in my environment (possibly because it has pandas installed) so I'm reverting this change for now.

Ah, of course. One of the few upsides of having a fresh Macbook.

[Fokko]

This is nice 👍

[Fokko]

This looks good to me. The V3 support can be added in a separate PR 👍

[Fokko]

Nit: we can move "downcast-ns-timestamp-to-us-on-write" into a constant, and reuse it in pyarrow.py

[sungwy]

Thank you for the review! I've adopted this in the new commits

[HonahX]

LGTM!