[DOP-21450] Add compressions for file formats

Author: Ilyas Gasanov

syncmaster/schemas/v1/transfers/file_format.py

@@ -22,8 +22,6 @@ class ORCCompression(str, Enum):
     NONE = "none"
     SNAPPY = "snappy"
     ZLIB = "zlib"
-    LZO = "lzo"
-    ZSTD = "zstd"
     LZ4 = "lz4"
 
 
@@ -32,10 +30,7 @@ class ParquetCompression(str, Enum):
     NONE = "none"
     SNAPPY = "snappy"
     GZIP = "gzip"
-    LZO = "lzo"
-    ZSTD = "zstd"
     LZ4 = "lz4"
-    BROTLI = "brotli"
 
 
 class JSONCompression(str, Enum):

tests/test_integration/test_run_transfer/test_hdfs.py

@@ -184,8 +184,8 @@ async def test_run_transfer_hdfs_to_postgres(
     )
     df = reader.run()
 
-    # as Excel does not support datetime values with precision greater than milliseconds
-    if file_format == "excel":
+    # as these file formats do not support datetime values with precision greater than milliseconds
+    if file_format in ("excel", "parquet", "orc"):
         df = df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
         init_df = init_df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
 
@@ -214,12 +214,12 @@ async def test_run_transfer_hdfs_to_postgres(
             id="excel",
         ),
         pytest.param(
-            ("orc", {"compression": "lzo"}),
+            ("orc", {"compression": "snappy"}),
             "with_compression",
             id="orc",
         ),
         pytest.param(
-            ("parquet", {"compression": "brotli"}),
+            ("parquet", {"compression": "lz4"}),
             "with_compression",
             id="parquet",
         ),
@@ -275,8 +275,8 @@ async def test_run_transfer_postgres_to_hdfs(
     )
     df = reader.run()
 
-    # as Excel does not support datetime values with precision greater than milliseconds
-    if format_name == "excel":
+    # as these file formats do not support datetime values with precision greater than milliseconds
+    if format_name in ("excel", "parquet", "orc"):
         init_df = init_df.withColumn(
             "REGISTERED_AT",
             to_timestamp(date_format(col("REGISTERED_AT"), "yyyy-MM-dd HH:mm:ss.SSS")),

tests/test_integration/test_run_transfer/test_s3.py

@@ -185,8 +185,8 @@ async def test_run_transfer_s3_to_postgres(
     )
     df = reader.run()
 
-    # as Excel does not support datetime values with precision greater than milliseconds
-    if file_format == "excel":
+    # as these file formats do not support datetime values with precision greater than milliseconds
+    if file_format in ("excel", "parquet", "orc"):
         df = df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
         init_df = init_df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
 
@@ -220,7 +220,7 @@ async def test_run_transfer_s3_to_postgres(
             id="orc",
         ),
         pytest.param(
-            ("parquet", {"compression": "brotli"}),
+            ("parquet", {"compression": "gzip"}),
             "with_compression",
             id="parquet",
         ),
@@ -276,8 +276,8 @@ async def test_run_transfer_postgres_to_s3(
     )
     df = reader.run()
 
-    # as Excel does not support datetime values with precision greater than milliseconds
-    if format_name == "excel":
+    # as these file formats do not support datetime values with precision greater than milliseconds
+    if format_name in ("excel", "parquet", "orc"):
         init_df = init_df.withColumn(
             "REGISTERED_AT",
             to_timestamp(date_format(col("REGISTERED_AT"), "yyyy-MM-dd HH:mm:ss.SSS")),

tests/test_unit/test_transfers/test_file_transfers/test_create_transfer.py

@@ -80,7 +80,7 @@
             "directory_path": "/some/parquet/path",
             "file_format": {
                 "type": "parquet",
-                "compression": "zstd",
+                "compression": "gzip",
             },
             "options": {
                 "some": "option",
@@ -175,7 +175,7 @@ async def test_developer_plus_can_create_s3_transfer(
         },
         "parquet": {
             "type": "parquet",
-            "compression": "zstd",
+            "compression": "gzip",
         },
     }
 
@@ -238,7 +238,7 @@ async def test_developer_plus_can_create_s3_transfer(
             "directory_path": "/some/parquet/path",
             "file_format": {
                 "type": "parquet",
-                "compression": "brotli",
+                "compression": "snappy",
             },
         },
     ],
@@ -331,7 +331,7 @@ async def test_developer_plus_can_create_hdfs_transfer(
         },
         "parquet": {
             "type": "parquet",
-            "compression": "brotli",
+            "compression": "snappy",
         },
     }