tests: integration: out_s3: add columnar format scenarios

Add out_s3 integration scenarios for the columnar formats: arrow and parquet with zstd, parquet and arrow with compression none, and a negative case asserting that an invalid format/compression combination (arrow + gzip) is rejected at startup. Each positive case verifies the ARROW1/PAR1 magic bytes and the absence of a Content-Encoding header.

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

Author: Rituparna Khaund

tests/integration/scenarios/out_s3/config/out_s3_arrow.yaml

@@ -0,0 +1,27 @@
+service:
+  flush: 1
+  grace: 3
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_s3
+      dummy: '{"message":"hello arrow s3","source":"dummy"}'
+      samples: 1
+
+  outputs:
+    - name: s3
+      match: out_s3
+      bucket: test-bucket
+      region: us-east-1
+      endpoint: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}
+      use_put_object: true
+      total_file_size: 1M
+      upload_timeout: 2s
+      s3_key_format: /payloads/$TAG/$UUID
+      format: arrow
+      compression: zstd
+      store_dir: /tmp/fluent-bit-test-suite-s3-arrow

tests/integration/scenarios/out_s3/config/out_s3_arrow_gzip_invalid.yaml

@@ -0,0 +1,27 @@
+service:
+  flush: 1
+  grace: 3
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_s3
+      dummy: '{"message":"hello arrow gzip s3","source":"dummy"}'
+      samples: 1
+
+  outputs:
+    - name: s3
+      match: out_s3
+      bucket: test-bucket
+      region: us-east-1
+      endpoint: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}
+      use_put_object: true
+      total_file_size: 1M
+      upload_timeout: 2s
+      s3_key_format: /payloads/$TAG/$UUID
+      format: arrow
+      compression: gzip
+      store_dir: /tmp/fluent-bit-test-suite-s3-arrow-gzip-invalid

tests/integration/scenarios/out_s3/config/out_s3_arrow_none.yaml

@@ -0,0 +1,27 @@
+service:
+  flush: 1
+  grace: 3
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_s3
+      dummy: '{"message":"hello arrow none s3","source":"dummy"}'
+      samples: 1
+
+  outputs:
+    - name: s3
+      match: out_s3
+      bucket: test-bucket
+      region: us-east-1
+      endpoint: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}
+      use_put_object: true
+      total_file_size: 1M
+      upload_timeout: 2s
+      s3_key_format: /payloads/$TAG/$UUID
+      format: arrow
+      compression: none
+      store_dir: /tmp/fluent-bit-test-suite-s3-arrow-none

tests/integration/scenarios/out_s3/config/out_s3_parquet.yaml

@@ -0,0 +1,27 @@
+service:
+  flush: 1
+  grace: 3
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_s3
+      dummy: '{"message":"hello parquet s3","source":"dummy"}'
+      samples: 1
+
+  outputs:
+    - name: s3
+      match: out_s3
+      bucket: test-bucket
+      region: us-east-1
+      endpoint: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}
+      use_put_object: true
+      total_file_size: 1M
+      upload_timeout: 2s
+      s3_key_format: /payloads/$TAG/$UUID
+      format: parquet
+      compression: zstd
+      store_dir: /tmp/fluent-bit-test-suite-s3-parquet

tests/integration/scenarios/out_s3/config/out_s3_parquet_none.yaml

@@ -0,0 +1,27 @@
+service:
+  flush: 1
+  grace: 3
+  log_level: info
+  http_server: on
+  http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}
+
+pipeline:
+  inputs:
+    - name: dummy
+      tag: out_s3
+      dummy: '{"message":"hello parquet none s3","source":"dummy"}'
+      samples: 1
+
+  outputs:
+    - name: s3
+      match: out_s3
+      bucket: test-bucket
+      region: us-east-1
+      endpoint: http://127.0.0.1:${TEST_SUITE_HTTP_PORT}
+      use_put_object: true
+      total_file_size: 1M
+      upload_timeout: 2s
+      s3_key_format: /payloads/$TAG/$UUID
+      format: parquet
+      compression: none
+      store_dir: /tmp/fluent-bit-test-suite-s3-parquet-none

tests/integration/scenarios/out_s3/tests/test_out_s3_001.py

@@ -140,6 +140,22 @@ def _start_or_skip_unsupported_s3_format(service, format_name):
         raise
 
 
+def _start_or_skip_unsupported_columnar_format(service, requires_marker):
+    """Start the service, skipping if the columnar format support
+    (arrow-glib/parquet-glib) was not compiled into the Fluent Bit binary."""
+    try:
+        service.start()
+    except FluentBitStartupError:
+        log_contents = ""
+        if service.service.flb and service.service.flb.log_file:
+            with open(service.service.flb.log_file, "r", encoding="utf-8", errors="replace") as file:
+                log_contents = file.read()
+        if requires_marker in log_contents or \
+                "unknown configuration property 'format'" in log_contents:
+            pytest.skip("columnar format support is not compiled into this Fluent Bit binary")
+        raise
+
+
 def test_out_s3_put_object_uploads_json_lines_payload():
     service = Service("out_s3_basic.yaml")
     service.start()
@@ -241,3 +257,74 @@ def test_out_s3_default_retry_exhausted_action_quarantines_file():
     service.stop()
 
     assert len(files) > 0
+
+
+def test_out_s3_format_arrow_uploads_feather_with_zstd():
+    service = Service("out_s3_arrow.yaml")
+    _start_or_skip_unsupported_columnar_format(service, "requires arrow-glib")
+    request = service.wait_for_request()
+    service.stop()
+
+    assert request["method"] == "PUT"
+    assert request["path"].startswith("/test-bucket/payloads/out_s3/")
+    body = request["body"]
+    # Arrow/Feather V2 files begin with the "ARROW1" magic. The object is the
+    # columnar file itself, so it must not carry a byte-level Content-Encoding.
+    assert body[:6] == b"ARROW1"
+    assert "Content-Encoding" not in request["headers"]
+
+
+def test_out_s3_format_parquet_uploads_parquet_with_zstd():
+    service = Service("out_s3_parquet.yaml")
+    _start_or_skip_unsupported_columnar_format(service, "requires parquet-glib")
+    request = service.wait_for_request()
+    service.stop()
+
+    assert request["method"] == "PUT"
+    assert request["path"].startswith("/test-bucket/payloads/out_s3/")
+    body = request["body"]
+    # Parquet files start and end with the "PAR1" magic. Page-level zstd is
+    # applied inside the file, so no byte-level Content-Encoding is expected.
+    assert body[:4] == b"PAR1"
+    assert body[-4:] == b"PAR1"
+    assert "Content-Encoding" not in request["headers"]
+
+
+def test_out_s3_format_parquet_compression_none_is_accepted():
+    # 'compression none' must be explicitly accepted (not rejected as an
+    # unknown codec) and produce an uncompressed Parquet object with no
+    # byte-level Content-Encoding header.
+    service = Service("out_s3_parquet_none.yaml")
+    _start_or_skip_unsupported_columnar_format(service, "requires parquet-glib")
+    request = service.wait_for_request()
+    service.stop()
+
+    assert request["method"] == "PUT"
+    body = request["body"]
+    assert body[:4] == b"PAR1"
+    assert body[-4:] == b"PAR1"
+    assert "Content-Encoding" not in request["headers"]
+
+
+def test_out_s3_format_arrow_compression_none_is_accepted():
+    # 'compression none' must be explicitly accepted (not rejected as an
+    # unknown codec) and produce an uncompressed Arrow/Feather object with no
+    # byte-level Content-Encoding header.
+    service = Service("out_s3_arrow_none.yaml")
+    _start_or_skip_unsupported_columnar_format(service, "requires arrow-glib")
+    request = service.wait_for_request()
+    service.stop()
+
+    assert request["method"] == "PUT"
+    body = request["body"]
+    assert body[:6] == b"ARROW1"
+    assert "Content-Encoding" not in request["headers"]
+
+
+def test_out_s3_format_arrow_compression_gzip_is_rejected():
+    # format=arrow with compression=gzip is an invalid combination;
+    # validate_format_compression() must reject it at plugin init.
+    service = Service("out_s3_arrow_gzip_invalid.yaml")
+    with pytest.raises(FluentBitStartupError):
+        service.start()
+    service.stop()