[Python] pandas to_parquet() writes Timestamps in a method unreadable by Redshift Spectrum

[IkeNefcy]

Describe the usage question you have. Please include as many useful details as possible.

Steps to reproduce:

1. Run a python script that will write timestamps to a parquet via s3.
2. Due to possible spectrum scan errors, each column should have the type defined. This means using df.['col1'] = pd.datetime(df.['col1'])
3. Once all types are assigned, Upload to s3, this can be via boto3 or by hand, it doesn't matter which.
4. From s3, perform a redshift copy (spectrum operation) (Table would be pre-created with a timestamp type column etc).
   example:

copy
    schema.test_table
from
    's3://bucket/test/test'
iam_role 'Spectrum_Role_neded'
FORMAT AS PARQUET;

5. Query the data, the timestamps appear corrupted.

From a deep dive it appears that an integer longer than the usual Unix timestamp is being written and this is causing the date to hundreds of thousands of years in the future.
For example a usual timestamp might look like 2023-09-11 10:39:00.000000
But the queried timestamp is 161065-10-07 20:24:18.134352 for the same uploaded timestamp.

This doesn't happen in Athena if you crawl the data and query via glue. Although, the crawler will be tricked as well and assume the type is "bigint".
In this case, for example
A normal timestamp if I change the schema to show that the bigint is really a time stamp, I can get 2023-10-02 00:00:00.000
The same number without this change, is 1696204800000000000, this looks similar to a unix timestamp, but with nano seconds.

It could be that nano seconds are part of the latest update and that aws redshift_spectrum is just not tracking on this change.
It would be nice to know if this is an intended change or not though, just opening this to confirm if this is expected behavior.
It's not super clear at a glance but I am seeing websites updated to recognize nano seconds. I'm wondering also if this is related to the epochalypse.

Component(s)

Python

[IkeNefcy]

Adding this is using Python 3.10.9, library is pyarrow 13.0.0, this does not happen with any version from 10.0.1 to 12.0.1

[mapleFU]

Would you mind upload a sample Parquet file or code, and report the error message here 🤔

By the way, I strongly suspect that it's because #36137 . Would you mind explicit set version='2.4' when calling to_parquet?

[IkeNefcy]

I'll grab a sample in a moment yeah. For error message there is no error, from upload to query there is no push back from any systems, it allows the parquet to be read during the copy function with no issue.
Also adding to clarify why we know it's Spectrum Specifically:
We tested downloading and rereading the files in pandas and in online parquet readers.
We tested on a cross account, and on the same account redshift was on.
We tested Lake Formation Hive Style tables, these show as external tables in Redshift and Spectrum reads these. For the table we tested, we can currently query in Athena and there is no issue, then the same query in Redshift is corrupted (this is not copying files, Spectrum is just reading from the s3 directly)

And I'll test out that versioning once I upload a sample.

[IkeNefcy]

This is how I am writing this sample

import pandas as pd
import boto3


def main():
    s3 = boto3.resource('s3')
    schedule = []
    row = {
        'start_time_local':'2023-09-11 10:39',
        'end_time_local':'2023-09-11 10:39',
        'start_time_utc':'2023-09-11 10:39',
        'end_time_utc':'2023-09-11 10:39'
    }
    schedule.append(row)
    data = pd.DataFrame(schedule)
    s3Loc = 'adhoc_tables/test'
    data['start_time_local'] = pd.to_datetime(data['start_time_local'])
    data['end_time_local'] = pd.to_datetime(data['end_time_local'])
    data['start_time_utc'] = pd.to_datetime(data['start_time_utc'])
    data['end_time_utc'] = pd.to_datetime(data['end_time_utc'])
    bucket = s3.Object('bucket_name', s3Loc)
    bucket.put(Body=data.to_parquet(index=False))
    return


if __name__ == "__main__":
    main()

Times are random. If I force pip to install <=12.0.1 this work fine. Then 13.0.0 it does not.

test.zip
This is a zip, it wouldn't allow me to upload a parquet

[mapleFU]

{
  "Version": "2.6",
  "CreatedBy": "parquet-cpp-arrow version 12.0.0",
  "TotalRows": "1",
  "NumberOfRowGroups": "1",
  "NumberOfRealColumns": "4",
  "NumberOfColumns": "4",
  "Columns": [
     { "Id": "0", "Name": "start_time_local", "PhysicalType": "INT64", "ConvertedType": "NONE", "LogicalType": {"Type": "Timestamp", "isAdjustedToUTC": false, "timeUnit": "microseconds", "is_from_converted_type": false, "force_set_converted_type": false} },
     { "Id": "1", "Name": "end_time_local", "PhysicalType": "INT64", "ConvertedType": "NONE", "LogicalType": {"Type": "Timestamp", "isAdjustedToUTC": false, "timeUnit": "microseconds", "is_from_converted_type": false, "force_set_converted_type": false} },
     { "Id": "2", "Name": "start_time_utc", "PhysicalType": "INT64", "ConvertedType": "NONE", "LogicalType": {"Type": "Timestamp", "isAdjustedToUTC": false, "timeUnit": "microseconds", "is_from_converted_type": false, "force_set_converted_type": false} },
     { "Id": "3", "Name": "end_time_utc", "PhysicalType": "INT64", "ConvertedType": "NONE", "LogicalType": {"Type": "Timestamp", "isAdjustedToUTC": false, "timeUnit": "microseconds", "is_from_converted_type": false, "force_set_converted_type": false} }
  ],
  "RowGroups": [
     {
       "Id": "0",  "TotalBytes": "368",  "TotalCompressedBytes": "384",  "Rows": "1",
       "ColumnChunks": [
          {"Id": "0", "Values": "1", "StatsSet": "True", "Stats": {"NumNulls": "0", "Max": "1694428740000000", "Min": "1694428740000000" },
           "Compression": "SNAPPY", "Encodings": "PLAIN(DICT_PAGE) RLE_DICTIONARY", "UncompressedSize": "92", "CompressedSize": "96" },
          {"Id": "1", "Values": "1", "StatsSet": "True", "Stats": {"NumNulls": "0", "Max": "1694428740000000", "Min": "1694428740000000" },
           "Compression": "SNAPPY", "Encodings": "PLAIN(DICT_PAGE) RLE_DICTIONARY", "UncompressedSize": "92", "CompressedSize": "96" },
          {"Id": "2", "Values": "1", "StatsSet": "True", "Stats": {"NumNulls": "0", "Max": "1694428740000000", "Min": "1694428740000000" },
           "Compression": "SNAPPY", "Encodings": "PLAIN(DICT_PAGE) RLE_DICTIONARY", "UncompressedSize": "92", "CompressedSize": "96" },
          {"Id": "3", "Values": "1", "StatsSet": "True", "Stats": {"NumNulls": "0", "Max": "1694428740000000", "Min": "1694428740000000" },
           "Compression": "SNAPPY", "Encodings": "PLAIN(DICT_PAGE) RLE_DICTIONARY", "UncompressedSize": "92", "CompressedSize": "96" }
        ]
     }
  ]
}

Oops, seems it's created by format 2.6 . The logic type is weird. Maybe we need check this logical type here.

cc @jorisvandenbossche I'm not so familiar with pandas, how can user switch to use parquet version 2.4 via that?

[IkeNefcy]

I will point out that by chance we found a fix last week when we didn't think this was a real issue, we thought it was related to the data we were uploading at the time.
We would change

data.to_parquet(index=False)

to

data.to_parquet(index=False, use_deprecated_int96_timestamps=True, flavor='spark')

And this works no issues

[IkeNefcy]

I'm not seeing any options for versions here other than kwargs https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.to_parquet.html. But I will upgrade back to 13.0.0 and test

[IkeNefcy]

Sorry for delay. I was able to test

data.to_parquet(index=False, version='2.4')

With no error. @mapleFU can I ask how you were able to view that metadata including the version? Would be helpful for testing. I'm skimming the binary but not seeing it.

[IkeNefcy]

nvm I was able to print something similar, but using the kwarg above for version it did not change the format used.

<pyarrow._parquet.FileMetaData object at 0xffff7c1dad90>
  created_by: parquet-cpp-arrow version 13.0.0
  num_columns: 4
  num_rows: 1
  num_row_groups: 1
  format_version: 2.6
  serialized_size: 2754

[jorisvandenbossche]

@IkeNefcy the file you uploaded (#38000 (comment)) is created with pyarrow 12.0, and is the file that works fine, is that correct?
So for that one, it is expected the timestamps are stored as microseconds, and this should work almost anywhere.

As @mapleFU mentioned, with #36137, we changed the default to start writing nanoseconds (if your original data is in nanoseconds, which is the case when starting from pandas) with pyarrow 13.0.

I assume that the parquet reader you are using with Spectrum is incorrectly reading those files. That's something best reported to them.

You can check the metadata of the Parquet file that was written using pyarrow as follows (using this with the file you uploaded):

In [17]: import pyarrow.parquet as pq

In [18]: meta = pq.read_metadata("Downloads/test")

In [19]: meta
Out[19]: 
<pyarrow._parquet.FileMetaData object at 0x7f4546eeae30>
  created_by: parquet-cpp-arrow version 12.0.0
  num_columns: 4
  num_rows: 1
  num_row_groups: 1
  format_version: 2.6
  serialized_size: 2754

In [20]: meta.schema
Out[20]: 
<pyarrow._parquet.ParquetSchema object at 0x7f4546e94500>
required group field_id=-1 schema {
  optional int64 field_id=-1 start_time_local (Timestamp(isAdjustedToUTC=false, timeUnit=microseconds, is_from_converted_type=false, force_set_converted_type=false));
  optional int64 field_id=-1 end_time_local (Timestamp(isAdjustedToUTC=false, timeUnit=microseconds, is_from_converted_type=false, force_set_converted_type=false));
  optional int64 field_id=-1 start_time_utc (Timestamp(isAdjustedToUTC=false, timeUnit=microseconds, is_from_converted_type=false, force_set_converted_type=false));
  optional int64 field_id=-1 end_time_utc (Timestamp(isAdjustedToUTC=false, timeUnit=microseconds, is_from_converted_type=false, force_set_converted_type=false));
}

(and so you can see here that this file was created using pyarrow 12.0, and created timestamps with microseconds)

[IkeNefcy]

Hey thanks for the reply, I was able to conclude the same thing about this falling on Spectrum and I let them know that there was a read issue. Good to know this is the default behavior in the newest version! I'll close this out since this answers all my questions. Ty again.