[DOP-21450] Add compression options for file formats except xml

Author: Ilyas Gasanov

docs/changelog/next_release/159.feature.rst

@@ -1 +1 @@
-Add compression options to file formats
+Add compression options to file formats CSV, JSON, JSONLine, Excel, ORC, Parquet

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"
@@ -51,13 +49,6 @@ class CSVCompression(str, Enum):
     DEFLATE = "deflate"
 
 
-class XMLCompression(str, Enum):
-    BZIP2 = "bzip2"
-    GZIP = "gzip"
-    LZ4 = "lz4"
-    SNAPPY = "snappy"
-
-
 class CSV(BaseModel):
     type: CSV_FORMAT
     delimiter: str = ","
@@ -66,21 +57,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 +84,13 @@ class XML(BaseModel):
     type: XML_FORMAT
     root_tag: str
     row_tag: str
-    compression: XMLCompression | None = None
 
 
 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

@@ -59,7 +59,6 @@
                 "type": "xml",
                 "root_tag": "data",
                 "row_tag": "record",
-                "compression": "lz4",
             },
             "options": {
                 "some": "option",
@@ -167,11 +166,10 @@ async def test_developer_plus_can_create_s3_transfer(
             "type": "xml",
             "root_tag": "data",
             "row_tag": "record",
-            "compression": "lz4",
         },
         "orc": {
             "type": "orc",
-            "compression": "none",
+            "compression": "zlib",
         },
         "parquet": {
             "type": "parquet",
@@ -223,7 +221,6 @@ async def test_developer_plus_can_create_s3_transfer(
                 "type": "xml",
                 "root_tag": "data",
                 "row_tag": "record",
-                "compression": "bzip2",
             },
         },
         {
@@ -323,11 +320,10 @@ async def test_developer_plus_can_create_hdfs_transfer(
             "type": "xml",
             "root_tag": "data",
             "row_tag": "record",
-            "compression": "bzip2",
         },
         "orc": {
             "type": "orc",
-            "compression": "none",
+            "compression": "zlib",
         },
         "parquet": {
             "type": "parquet",