refactor: make benchmark result clearly

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

@@ -0,0 +1,148 @@
+/*
+ * Copyright 2023 Greptime Team
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.greptime.bench;
+
+import io.greptime.common.util.Cpus;
+import io.greptime.common.util.SystemPropertyUtil;
+import io.greptime.models.TableSchema;
+import org.slf4j.Logger;
+
+/**
+ * Utility class for printing benchmark results in a consistent format.
+ */
+public class BenchmarkResultPrinter {
+
+    public static void printBenchmarkHeader(Logger log, String apiType) {
+        log.info("=== GreptimeDB {} API Log Benchmark ===", apiType);
+        log.info("Synthetic log data generation and {} API ingestion performance test", apiType.toLowerCase());
+        log.info("");
+    }
+
+    public static void printConfiguration(
+            Logger log, String apiType, boolean zstdCompression, int batchSize, int parallelismOrConcurrency) {
+        printConfiguration(log, apiType, zstdCompression, batchSize, parallelismOrConcurrency, null);
+    }
+
+    public static void printConfiguration(
+            Logger log,
+            String apiType,
+            boolean zstdCompression,
+            int batchSize,
+            int parallelismOrConcurrency,
+            Integer maxPointsPerSecond) {
+        log.info("=== {} API Benchmark Configuration ===", apiType);
+
+        String endpointsStr = SystemPropertyUtil.get("db.endpoints");
+        if (endpointsStr == null) {
+            endpointsStr = "localhost:4001";
+        }
+        String database = SystemPropertyUtil.get("db.database");
+        if (database == null) {
+            database = "public";
+        }
+
+        log.info("Endpoint: {}", endpointsStr);
+        log.info("Database: {}", database);
+        log.info("Batch size: {}", batchSize);
+
+        if (maxPointsPerSecond != null) {
+            log.info(
+                    "Max points per second: {}",
+                    maxPointsPerSecond == Integer.MAX_VALUE ? "unlimited" : maxPointsPerSecond);
+        } else if (apiType.equals("Bulk")) {
+            log.info("Parallelism: {}", parallelismOrConcurrency);
+        } else {
+            log.info("Concurrency: {}", parallelismOrConcurrency);
+        }
+
+        log.info("Compression: {}", (zstdCompression ? "zstd" : "none"));
+        log.info("CPU cores: {}", Cpus.cpus());
+        log.info("Build profile: release");
+        log.info("");
+    }
+
+    public static void printBenchmarkStart(
+            Logger log,
+            String apiType,
+            TableDataProvider provider,
+            TableSchema schema,
+            int batchSize,
+            int parallelismOrConcurrency) {
+        printBenchmarkStart(log, apiType, provider, schema, batchSize, parallelismOrConcurrency, null);
+    }
+
+    public static void printBenchmarkStart(
+            Logger log,
+            String apiType,
+            TableDataProvider provider,
+            TableSchema schema,
+            int batchSize,
+            int parallelismOrConcurrency,
+            Integer maxPointsPerSecond) {
+        log.info("=== Running {} API Log Data Benchmark ===", apiType);
+        log.info("Setting up {} writer...", apiType.toLowerCase());
+        log.info(
+                "Starting {} API benchmark: {}",
+                apiType.toLowerCase(),
+                provider.getClass().getSimpleName());
+        log.info(
+                "Table: {} ({} columns)",
+                schema.getTableName(),
+                schema.getColumnNames().size());
+        log.info("Target rows: {}", provider.rowCount());
+        log.info("Batch size: {}", batchSize);
+
+        if (maxPointsPerSecond != null) {
+            log.info(
+                    "Max points per second: {}",
+                    maxPointsPerSecond == Integer.MAX_VALUE ? "unlimited" : maxPointsPerSecond);
+        } else if (apiType.equals("Bulk")) {
+            log.info("Parallelism: {}", parallelismOrConcurrency);
+        } else {
+            log.info("Concurrency: {}", parallelismOrConcurrency);
+        }
+
+        log.info("");
+    }
+
+    public static void printBatchProgress(Logger log, long batch, long totalRows, long writeRatePerSecond) {
+        log.info("→ Batch {}: {} rows processed ({} rows/sec)", batch, totalRows, writeRatePerSecond);
+
+        if (batch % 10 == 0) {
+            log.info("Flushed {} responses (total {} affected rows)", batch, totalRows);
+        }
+    }
+
+    public static void printCompletionMessages(Logger log, String apiType) {
+        log.info("Finishing {} writer and waiting for all responses...", apiType.toLowerCase());
+        log.info("All {} writes completed successfully", apiType.toLowerCase());
+        log.info("Cleaning up data provider...");
+        log.info("{} API benchmark completed successfully!", apiType);
+    }
+
+    public static void printBenchmarkSummary(
+            Logger log, TableDataProvider provider, long totalRows, long durationMs, long throughput) {
+        log.info("=== Benchmark Result ===");
+        log.info("Table: {}", provider.tableSchema().getTableName());
+        log.info("");
+        log.info("Provider                          Rows Duration(ms)      Throughput     Status");
+        log.info("--------------------------------------------------------------------------");
+        log.info(String.format(
+                "%-30s %8d %12d %12d r/s    SUCCESS",
+                provider.getClass().getSimpleName(), totalRows, durationMs, throughput));
+    }
+}

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

@@ -17,8 +17,10 @@
 package io.greptime.bench;
 
 import io.greptime.GreptimeDB;
+import io.greptime.common.util.SystemPropertyUtil;
 import io.greptime.options.GreptimeOptions;
 import io.greptime.quickstart.query.QueryJDBCQuickStart;
+import io.greptime.rpc.RpcOptions;
 import java.io.IOException;
 import java.util.Properties;
 import org.slf4j.Logger;
@@ -38,11 +40,23 @@ public static GreptimeDB connect() {
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
-        String database = (String) prop.get("db.database");
-        String endpointsStr = prop.getProperty("db.endpoints");
+
+        String database = SystemPropertyUtil.get("db.database");
+        if (database == null) {
+            database = (String) prop.get("db.database");
+        }
+
+        String endpointsStr = SystemPropertyUtil.get("db.endpoints");
+        if (endpointsStr == null) {
+            endpointsStr = prop.getProperty("db.endpoints");
+        }
+
         String[] endpoints = endpointsStr.split(",");
+        RpcOptions rpcOptions = RpcOptions.newDefault();
+        rpcOptions.setDefaultRpcTimeout(60 * 1000);
         GreptimeOptions opts = GreptimeOptions.newBuilder(endpoints, database)
                 .writeMaxRetries(0)
+                .rpcOptions(rpcOptions)
                 .defaultStreamMaxWritePointsPerSecond(Integer.MAX_VALUE)
                 .maxInFlightWritePoints(Integer.MAX_VALUE)
                 .useZeroCopyWriteInBulkWrite(true)

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

@@ -37,14 +37,12 @@
 public class MultiProducerTableDataProvider extends RandomTableDataProvider {
 
     private final int producerCount;
-    private final long rowCount;
     private final ExecutorService executorService;
     private final BlockingQueue<Object[]> buffer = new ArrayBlockingQueue<>(100000);
 
     {
         this.producerCount = SystemPropertyUtil.getInt("multi_producer_table_data_provider.producer_count", 10);
         // Total number of rows to generate, configurable via system property
-        this.rowCount = SystemPropertyUtil.getLong("table_data_provider.row_count", 10_000_000L);
         this.executorService = ThreadPoolUtil.newBuilder()
                 .poolName("multi-producer-table-data-provider")
                 .enableMetric(true)
@@ -62,7 +60,7 @@ public void init() {
         AtomicLong rowIndex = new AtomicLong(0);
         for (int i = 0; i < producerCount; i++) {
             this.executorService.execute(() -> {
-                while (rowIndex.getAndIncrement() < rowCount) {
+                while (rowIndex.getAndIncrement() < rowCount()) {
                     Object[] row = nextRow();
                     try {
                         buffer.put(row);
@@ -82,7 +80,7 @@ public Iterator<Object[]> rows() {
 
             @Override
             public boolean hasNext() {
-                return index < rowCount;
+                return index < rowCount();
             }
 
             @Override

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

@@ -51,7 +51,7 @@ public class RandomTableDataProvider implements TableDataProvider {
                 .addField("pod_name", DataType.String)
                 .build();
         // Total number of rows to generate, configurable via system property
-        rowCount = SystemPropertyUtil.getLong("table_data_provider.row_count", 10_000_000L);
+        rowCount = SystemPropertyUtil.getLong("table_data_provider.row_count", 5_000_000L);
     }
 
     @Override
@@ -133,4 +133,9 @@ public Object[] next() {
 
     @Override
     public void close() throws Exception {}
+
+    @Override
+    public long rowCount() {
+        return rowCount;
+    }
 }

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

@@ -38,4 +38,9 @@ public interface TableDataProvider extends AutoCloseable {
      * Returns the iterator of the rows.
      */
     Iterator<Object[]> rows();
+
+    /**
+     * Returns the total number of rows.
+     */
+    long rowCount();
 }