perf: optimize Arrow-to-Parquet conversion and sync scheduling (#1784)

- add ordered forward Arrow IPC reading for metric conversion
- decode multiple small files and multi-batch files concurrently with bounded reader lanes
- pipeline metric row-group preparation and Parquet encoding while preserving output order
- parallelize metric sorting and column encoding
- add detailed Arrow-to-Parquet phase timings
- bound periodic Parquet group concurrency per stream
- allow configurable overlapping periodic local-sync cycles
- atomically isolate processing directories used by concurrent cycles
- serialize object-store sync and retain upload claims until staged files are removed
- refresh retained claims after deletion failures to prevent duplicate uploads
- avoid unnecessary schema cloning during batch adaptation
- document conversion sizing and concurrency formulas
- add coverage for reader ordering, dictionary batches, invalid files, and Parquet row-group correctness

Configuration:

New environment variables:
- MAX_PERIODIC_LOCAL_SYNC_CYCLES (default: 2):
maximum periodic local-sync cycles allowed to run concurrently

- METRIC_ARROW_READERS_PER_FILE (default: 4):
maximum parallel readers used for record batches within one metric Arrow file
increase for multi-batch Arrow files when decoding is slow and CPU is available
decrease for one-batch files or CPU contention

- METRIC_ARROW_READERS_IN_FLIGHT (default: 4):
maximum total Arrow reader lanes active per metric Parquet conversion
increase when many small Arrow files must be decoded and CPU/I/O has headroom
decrease when many datasets convert concurrently or CPU is saturated

- METRIC_ROW_GROUP_ENCODE_IN_FLIGHT (default: 2)
maximum metric row groups encoded concurrently per Parquet file
increase when Parquet encoding dominates and CPU/memory is available
decrease for wide schemas, high memory usage, or many concurrent conversions

- PARQUET_COLUMNS_PER_RAYON_TASK (default: 8)
minimum Parquet columns assigned to each Rayon encoding task
decrease to expose more column parallelism
increase for wide schemas or many concurrent conversions to reduce task-scheduling overhead

Author: nikhilsinhaparseable

docker-compose-distributed-test-with-kafka.yaml

@@ -30,7 +30,7 @@ services:
 
   # minio
   minio:
-    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
+    image: quay.io/parseablehq/minio:edge
     entrypoint:
       - sh
       - -euc

docker-compose-distributed-test.yaml

@@ -4,7 +4,7 @@ networks:
 services:
   # minio
   minio:
-    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
+    image: quay.io/parseablehq/minio:edge
     entrypoint:
       - sh
       - -euc
@@ -83,6 +83,7 @@ services:
       - P_PARQUET_COMPRESSION_ALGO=snappy
       - P_MODE=ingest
       - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000
+      - MAX_PERIODIC_LOCAL_SYNC_CYCLES=1
       - RUST_LOG=warn
     networks:
       - parseable-internal

docker-compose-test-with-kafka.yaml

@@ -3,7 +3,7 @@ networks:
 
 services:
   minio:
-    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
+    image: quay.io/parseablehq/minio:edge
     entrypoint:
       - sh
       - -euc

docker-compose-test.yaml

@@ -3,7 +3,7 @@ networks:
 
 services:
   minio:
-    image: minio/minio:RELEASE.2025-02-03T21-03-04Z
+    image: quay.io/parseablehq/minio:edge
     entrypoint:
       - sh
       - -euc
@@ -43,6 +43,7 @@ services:
       - P_PASSWORD=parseableadmin
       - P_CHECK_UPDATE=false
       - P_PARQUET_COMPRESSION_ALGO=snappy
+      - MAX_PERIODIC_LOCAL_SYNC_CYCLES=1
       - RUST_LOG=warn
     depends_on:
       minio:

src/handlers/livetail.rs

@@ -128,7 +128,7 @@ impl FlightService for FlightServiceImpl {
 
         let adapter_schema = schema.clone();
         let rx = rx.map(move |x| match x {
-            Message::Record(t) => Ok(utils::arrow::adapt_batch(&adapter_schema, &t)),
+            Message::Record(t) => Ok(utils::arrow::adapt_batch(adapter_schema.clone(), &t)),
             Message::Skipped(_) => {
                 warn!("livetail channel capacity is full.");
                 Ok(RecordBatch::new_empty(adapter_schema.clone()))