fix: Allow trancated timestamps when converting (#3861)

feast-dev · Jan 19, 2024 · bdd7dfb · bdd7dfb
1 parent 8f65fe1
commit bdd7dfb
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 3 deletions.
diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py
@@ -356,7 +356,10 @@ def write_logged_features(
             # In Pyarrow v13.0, the parquet version was upgraded to v2.6 from v2.4.
             # Set the coerce_timestamps to "us"(microseconds) for backward compatibility.
             pyarrow.parquet.write_table(
-                table=data, where=parquet_temp_file, coerce_timestamps="us"
+                table=data,
+                where=parquet_temp_file,
+                coerce_timestamps="us",
+                allow_truncated_timestamps=True,
             )
 
             parquet_temp_file.seek(0)
@@ -407,7 +410,10 @@ def offline_write_batch(
             # In Pyarrow v13.0, the parquet version was upgraded to v2.6 from v2.4.
             # Set the coerce_timestamps to "us"(microseconds) for backward compatibility.
             pyarrow.parquet.write_table(
-                table=table, where=parquet_temp_file, coerce_timestamps="us"
+                table=table,
+                where=parquet_temp_file,
+                coerce_timestamps="us",
+                allow_truncated_timestamps=True,
             )
 
             parquet_temp_file.seek(0)

diff --git a/sdk/python/feast/infra/utils/aws_utils.py b/sdk/python/feast/infra/utils/aws_utils.py
@@ -353,7 +353,12 @@ def upload_arrow_table_to_redshift(
         with tempfile.TemporaryFile(suffix=".parquet") as parquet_temp_file:
             # In Pyarrow v13.0, the parquet version was upgraded to v2.6 from v2.4.
             # Set the coerce_timestamps to "us"(microseconds) for backward compatibility.
-            pq.write_table(table, parquet_temp_file, coerce_timestamps="us")
+            pq.write_table(
+                table,
+                parquet_temp_file,
+                coerce_timestamps="us",
+                allow_truncated_timestamps=True,
+            )
             parquet_temp_file.seek(0)
             s3_resource.Object(bucket, key).put(Body=parquet_temp_file)