fix: Allow trancated timestamps when converting (feast-dev#3861)

j1wonpark · web-flow · commit bdd7dfb6128d · 2024-01-18T23:17:41.000-05:00
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)
 
