chore: minor change

Author: fengjiachun

ingester-bulk-protocol/src/main/java/io/greptime/BulkWriteService.java

@@ -61,7 +61,7 @@ public class BulkWriteService implements AutoCloseable {
     private final ClientStreamListener listener;
     private final AsyncPutListener metadataListener;
     private final long timeoutMs;
-    
+
     /**
      * Constructs a new BulkWriteService.
      *

ingester-bulk-protocol/src/main/java/org/apache/arrow/flight/BulkFlightClient.java

@@ -205,10 +205,7 @@ private static class OnStreamReadyHandler implements Runnable {
 
         @Override
         public void run() {
-            int mayReleasePermits = this.maxRequestsInFlight - this.semaphore.availablePermits();
-            if (mayReleasePermits > 0) {
-                this.semaphore.release(mayReleasePermits);
-            }
+            this.semaphore.release(this.maxRequestsInFlight);
         }
 
         /**

ingester-example/src/main/java/io/greptime/bench/BulkWriteBenchmark.java

@@ -58,7 +58,7 @@ public static void main(String[] args) throws Exception {
                 .allocatorInitReservation(0)
                 .allocatorMaxAllocation(4 * 1024 * 1024 * 1024L)
                 .timeoutMsPerMessage(60000)
-                .maxRequestsInFlight(4)
+                .maxRequestsInFlight(8)
                 .build();
         Compression compression = zstdCompression ? Compression.Zstd : Compression.None;
         Context ctx = Context.newDefault().withCompression(compression);

ingester-example/src/main/java/io/greptime/bench/MultiProducerTableDataProvider.java

@@ -30,16 +30,12 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicLong;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 @SPI(
         name = "multi_producer_table_data_provider",
         priority = 10 /* newer implementation can use higher priority to override the old one */)
 public class MultiProducerTableDataProvider extends RandomTableDataProvider {
 
-    private static final Logger LOG = LoggerFactory.getLogger(MultiProducerTableDataProvider.class);
-
     private final int producerCount;
     private final long rowCount;
     private final ExecutorService executorService;
@@ -91,17 +87,13 @@ public boolean hasNext() {
 
             @Override
             public Object[] next() {
-                index++;
-                Object[] row = buffer.poll();
-                if (row == null) {
-                    try {
-                        LOG.info("Waiting for row from buffer");
-                        row = buffer.take();
-                    } catch (InterruptedException e) {
-                        throw new RuntimeException(e);
-                    }
+                try {
+                    Object[] row = buffer.take();
+                    index++;
+                    return row;
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
                 }
-                return row;
             }
         };
     }

ingester-example/src/main/java/io/greptime/bench/RandomTableDataProvider.java

@@ -36,9 +36,9 @@ public class RandomTableDataProvider implements TableDataProvider {
     {
         tableSchema = TableSchema.newBuilder("my_bench_table")
                 .addTimestamp("log_ts", DataType.TimestampMillisecond)
-                .addTag("business_name", DataType.String)
-                .addTag("app_name", DataType.String)
-                .addTag("host_name", DataType.String)
+                .addField("business_name", DataType.String)
+                .addField("app_name", DataType.String)
+                .addField("host_name", DataType.String)
                 .addField("log_message", DataType.String) // 2K
                 .addField("log_level", DataType.String)
                 .addField("log_name", DataType.String)

ingester-protocol/src/main/java/io/greptime/BulkWrite.java

@@ -45,7 +45,7 @@ public interface BulkWrite {
      * This value should be determined based on the size of each request packet. A higher value means more in-flight requests,
      * which could potentially saturate network bandwidth or exceed the actual processing capacity of the database.
      */
-    int DEFAULT_MAX_REQUESTS_IN_FLIGHT = 4;
+    int DEFAULT_MAX_REQUESTS_IN_FLIGHT = 8;
 
     static class Config {
         private long allocatorInitReservation = DEFAULT_ALLOCATOR_INIT_RESERVATION;