tests: internal: aws_compress: add parquet page-level compression tests

Add unit tests for out_s3_compress_parquet covering snappy, zstd, gzip
and uncompressed page-level codecs, verifying the PAR1 magic markers and
that compression reduces the output size for repetitive input.

Signed-off-by: Rituparna Khaund <ritukhau@amazon.co.uk>

Author: Rituparna Khaund

tests/internal/aws_compress.c

@@ -105,7 +105,7 @@ void test_compression_snappy_return_value_normalization()
 {
     /* This test verifies that the snappy wrapper correctly normalizes return values
      * to conform to the AWS compression interface contract: -1 on error, 0 on success.
-     * 
+     *
      * The test uses the actual flb_aws_compression_compress function which internally
      * uses the wrapper. We verify that successful compression returns exactly 0,
      * demonstrating that the wrapper properly normalizes the return value.
@@ -115,17 +115,17 @@ void test_compression_snappy_return_value_normalization()
     size_t out_len = 0;
     int compression_type;
     char test_data[] = "test data for compression";
-    
+
     compression_type = flb_aws_compression_get_type("snappy");
     TEST_CHECK(compression_type != -1);
-    
+
     /* Test successful compression - should return exactly 0 (not any other value) */
-    ret = flb_aws_compression_compress(compression_type, test_data, 
+    ret = flb_aws_compression_compress(compression_type, test_data,
                                       strlen(test_data), &out_data, &out_len);
     TEST_CHECK(ret == 0);
     TEST_MSG("Expected return value 0 on success, got: %d", ret);
     TEST_MSG("This verifies the wrapper returns 0 (not passthrough of underlying function)");
-    
+
     if (ret == 0 && out_data != NULL) {
         TEST_CHECK(out_len > 0);
         TEST_MSG("Compressed data length: %zu", out_len);
@@ -334,6 +334,122 @@ void test_b64_truncated_gzip_boundary()
     flb_aws_compress_truncate_b64_test_cases__gzip_decode(cases, 40);
 }
 
+#ifdef FLB_HAVE_ARROW_PARQUET
+void test_parquet_format_snappy()
+{
+    int ret;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    char *json = "{\"key\":\"value\",\"num\":42}\n"
+                 "{\"key\":\"other\",\"num\":99}\n";
+    size_t json_len = strlen(json);
+
+    ret = out_s3_compress_parquet(json, json_len, &out_buf, &out_size,
+                                 FLB_PARQUET_COMPRESSION_SNAPPY);
+    if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) {
+        TEST_MSG("Parquet SNAPPY conversion failed");
+        return;
+    }
+    TEST_CHECK(memcmp(out_buf, "PAR1", 4) == 0);
+    TEST_CHECK(memcmp((char *)out_buf + out_size - 4, "PAR1", 4) == 0);
+    flb_free(out_buf);
+}
+
+void test_parquet_format_zstd()
+{
+    int ret;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    char *json = "{\"key\":\"value\",\"num\":42}\n"
+                 "{\"key\":\"other\",\"num\":99}\n";
+    size_t json_len = strlen(json);
+
+    ret = out_s3_compress_parquet(json, json_len, &out_buf, &out_size,
+                                 FLB_PARQUET_COMPRESSION_ZSTD);
+    if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) {
+        TEST_MSG("Parquet ZSTD conversion failed");
+        return;
+    }
+    TEST_CHECK(memcmp(out_buf, "PAR1", 4) == 0);
+    TEST_CHECK(memcmp((char *)out_buf + out_size - 4, "PAR1", 4) == 0);
+    flb_free(out_buf);
+}
+
+void test_parquet_format_gzip()
+{
+    int ret;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    char *json = "{\"key\":\"value\",\"num\":42}\n"
+                 "{\"key\":\"other\",\"num\":99}\n";
+    size_t json_len = strlen(json);
+
+    ret = out_s3_compress_parquet(json, json_len, &out_buf, &out_size,
+                                 FLB_PARQUET_COMPRESSION_GZIP);
+    if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) {
+        TEST_MSG("Parquet GZIP conversion failed");
+        return;
+    }
+    TEST_CHECK(memcmp(out_buf, "PAR1", 4) == 0);
+    TEST_CHECK(memcmp((char *)out_buf + out_size - 4, "PAR1", 4) == 0);
+    flb_free(out_buf);
+}
+
+void test_parquet_format_uncompressed()
+{
+    int ret;
+    void *out_buf = NULL;
+    size_t out_size = 0;
+    char *json = "{\"key\":\"value\",\"num\":42}\n"
+                 "{\"key\":\"other\",\"num\":99}\n";
+    size_t json_len = strlen(json);
+
+    ret = out_s3_compress_parquet(json, json_len, &out_buf, &out_size,
+                                 FLB_PARQUET_COMPRESSION_NONE);
+    if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) {
+        TEST_MSG("Parquet NONE conversion failed");
+        return;
+    }
+    TEST_CHECK(memcmp(out_buf, "PAR1", 4) == 0);
+    TEST_CHECK(memcmp((char *)out_buf + out_size - 4, "PAR1", 4) == 0);
+    flb_free(out_buf);
+}
+
+void test_parquet_compression_reduces_size()
+{
+    int ret;
+    void *buf_none = NULL;
+    void *buf_snappy = NULL;
+    size_t size_none = 0;
+    size_t size_snappy = 0;
+    char *json = "{\"msg\":\"hello hello hello hello hello hello\"}\n"
+                 "{\"msg\":\"hello hello hello hello hello hello\"}\n"
+                 "{\"msg\":\"hello hello hello hello hello hello\"}\n"
+                 "{\"msg\":\"hello hello hello hello hello hello\"}\n"
+                 "{\"msg\":\"hello hello hello hello hello hello\"}\n";
+    size_t json_len = strlen(json);
+
+    ret = out_s3_compress_parquet(json, json_len, &buf_none, &size_none,
+                                 FLB_PARQUET_COMPRESSION_NONE);
+    if (!TEST_CHECK(ret == 0 && buf_none != NULL)) {
+        TEST_MSG("Parquet NONE conversion failed");
+        return;
+    }
+
+    ret = out_s3_compress_parquet(json, json_len, &buf_snappy, &size_snappy,
+                                 FLB_PARQUET_COMPRESSION_SNAPPY);
+    if (!TEST_CHECK(ret == 0 && buf_snappy != NULL)) {
+        TEST_MSG("Parquet SNAPPY conversion failed");
+        flb_free(buf_none);
+        return;
+    }
+    TEST_CHECK(size_snappy <= size_none);
+
+    flb_free(buf_none);
+    flb_free(buf_snappy);
+}
+#endif
+
 TEST_LIST = {
     { "test_compression_gzip", test_compression_gzip },
     { "test_compression_zstd", test_compression_zstd },
@@ -352,6 +468,14 @@ TEST_LIST = {
       test_b64_truncated_gzip_truncation_multi_rounds },
     { "test_b64_truncated_gzip_boundary",
       test_b64_truncated_gzip_boundary },
+#ifdef FLB_HAVE_ARROW_PARQUET
+    { "test_parquet_format_snappy", test_parquet_format_snappy },
+    { "test_parquet_format_zstd", test_parquet_format_zstd },
+    { "test_parquet_format_gzip", test_parquet_format_gzip },
+    { "test_parquet_format_uncompressed", test_parquet_format_uncompressed },
+    { "test_parquet_compression_reduces_size",
+      test_parquet_compression_reduces_size },
+#endif
     { 0 }
 };
 
@@ -419,8 +543,8 @@ static void flb_aws_compress_general_test_cases(int test_type,
     while (tcase->compression_keyword != 0) {
 
         size_t in_data_len = strlen(tcase->in_data);
-        compression_type = flb_aws_compression_get_type(tcase->compression_keyword);   
-        
+        compression_type = flb_aws_compression_get_type(tcase->compression_keyword);
+
         TEST_CHECK(compression_type != -1);
         TEST_MSG("| flb_aws_get_compression_type: failed to get compression type for "
                  "keyword "