[DOP-21450] Add compressions for file formats

Author: Ilyas Gasanov

syncmaster/schemas/v1/transfers/file_format.py

@@ -18,15 +18,13 @@
 
 
 class ORCCompression(str, Enum):
-    UNCOMPRESSED = "uncompressed"
     NONE = "none"
     SNAPPY = "snappy"
     ZLIB = "zlib"
     LZ4 = "lz4"
 
 
 class ParquetCompression(str, Enum):
-    UNCOMPRESSED = "uncompressed"
     NONE = "none"
     SNAPPY = "snappy"
     GZIP = "gzip"
@@ -52,6 +50,7 @@ class CSVCompression(str, Enum):
 
 
 class XMLCompression(str, Enum):
+    NONE = "none"
     BZIP2 = "bzip2"
     GZIP = "gzip"
     LZ4 = "lz4"
@@ -66,21 +65,21 @@ class CSV(BaseModel):
     escape: str = "\\"
     include_header: bool = False
     line_sep: str = "\n"
-    compression: CSVCompression = CSVCompression.NONE
+    compression: CSVCompression = CSVCompression.GZIP
 
 
 class JSONLine(BaseModel):
     type: JSONLINE_FORMAT
     encoding: str = "utf-8"
     line_sep: str = "\n"
-    compression: JSONCompression = CSVCompression.NONE
+    compression: JSONCompression = JSONCompression.GZIP
 
 
 class JSON(BaseModel):
     type: JSON_FORMAT
     encoding: str = "utf-8"
     line_sep: str = "\n"
-    compression: JSONCompression = CSVCompression.NONE
+    compression: JSONCompression = JSONCompression.GZIP
 
 
 class Excel(BaseModel):
@@ -93,14 +92,14 @@ class XML(BaseModel):
     type: XML_FORMAT
     root_tag: str
     row_tag: str
-    compression: XMLCompression | None = None
+    compression: XMLCompression = XMLCompression.GZIP
 
 
 class ORC(BaseModel):
     type: ORC_FORMAT
-    compression: ORCCompression = CSVCompression.NONE
+    compression: ORCCompression = ORCCompression.ZLIB
 
 
 class Parquet(BaseModel):
     type: PARQUET_FORMAT
-    compression: ParquetCompression = CSVCompression.NONE
+    compression: ParquetCompression = ParquetCompression.SNAPPY

tests/resources/file_df_connection/generate_files.py

@@ -64,7 +64,7 @@ def get_pyarrow_schema() -> ArrowSchema:
             pa.field("REGION", pa.string()),
             pa.field("NUMBER", pa.int32()),
             pa.field("BIRTH_DATE", pa.date32()),
-            pa.field("REGISTERED_AT", pa.timestamp("ms")),
+            pa.field("REGISTERED_AT", pa.timestamp("us")),
             pa.field("ACCOUNT_BALANCE", pa.float64()),
         ],
     )

tests/resources/file_df_connection/orc/with_compression/file.snappy.orc

@@ -185,7 +185,7 @@ 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 in ("excel", "parquet", "orc"):
+    if file_format == "excel":
         df = df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
         init_df = init_df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
 
@@ -276,7 +276,7 @@ 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 in ("excel", "parquet"):
+    if format_name == "excel":
         init_df = init_df.withColumn(
             "REGISTERED_AT",
             to_timestamp(date_format(col("REGISTERED_AT"), "yyyy-MM-dd HH:mm:ss.SSS")),

tests/resources/file_df_connection/orc/without_compression/file.orc

@@ -186,7 +186,7 @@ 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 in ("excel", "parquet", "orc"):
+    if file_format == "excel":
         df = df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
         init_df = init_df.withColumn("REGISTERED_AT", date_trunc("second", col("REGISTERED_AT")))
 
@@ -277,7 +277,7 @@ 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 in ("excel", "parquet"):
+    if format_name == "excel":
         init_df = init_df.withColumn(
             "REGISTERED_AT",
             to_timestamp(date_format(col("REGISTERED_AT"), "yyyy-MM-dd HH:mm:ss.SSS")),

tests/resources/file_df_connection/parquet/with_compression/file.snappy.parquet

@@ -171,7 +171,7 @@ async def test_developer_plus_can_create_s3_transfer(
         },
         "orc": {
             "type": "orc",
-            "compression": "none",
+            "compression": "zlib",
         },
         "parquet": {
             "type": "parquet",
@@ -327,7 +327,7 @@ async def test_developer_plus_can_create_hdfs_transfer(
         },
         "orc": {
             "type": "orc",
-            "compression": "none",
+            "compression": "zlib",
         },
         "parquet": {
             "type": "parquet",