feat: bulk write pipeline (#68)

* feat: remove common auth method

* feat: put result

* feat: refactor write api

* feat: write pipeline

* refactor: big refactor of bulk write impl

* chore: avoid unnecessary busy-wait on stream ready

* chore: by review

Author: fengjiachun

ingester-bulk-protocol/pom.xml

@@ -77,6 +77,10 @@
             <groupId>com.google.protobuf</groupId>
             <artifactId>protobuf-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
 
         <!-- test -->
         <dependency>

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

@@ -24,10 +24,12 @@ public enum ArrowCompressionType {
      * Zstd compression.
      */
     Zstd,
+
     /**
      * Lz4 compression.
      */
     Lz4,
+
     /**
      * No compression.
      */

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

@@ -23,7 +23,6 @@
 import io.greptime.common.util.MetricsUtil;
 import io.greptime.rpc.TlsOptions;
 import io.netty.util.internal.SystemPropertyUtil;
-import org.apache.arrow.flight.AsyncPutListener;
 import org.apache.arrow.flight.BulkFlightClient;
 import org.apache.arrow.flight.BulkFlightClient.ClientStreamListener;
 import org.apache.arrow.flight.BulkFlightClient.PutListener;
@@ -45,15 +44,9 @@
  * BulkWriteManager is a specialized manager for efficiently writing block data to the server.
  *
  * It encapsulates a Flight client and a buffer allocator to manage memory resources.
- *
  * The primary function of this manager is to establish bulk write streams,
  * which provide an optimized channel for transmitting block data to the server.
  * These streams handle the serialization and transfer of data in an efficient manner.
- *
- * <p>
- * `ZERO_COPY_WRITE` is disabled by default, if you want to enable it, you can set the system property
- * `arrow.flight.enable_zero_copy_write` to `true`.
- * </p>
  */
 public class BulkWriteManager implements AutoCloseable {
 
@@ -136,43 +129,26 @@ public static BulkWriteManager create(
         return client;
     }
 
-    /**
-     * @see #intoBulkWriteStream(String, String, Schema, long, CallOption...)
-     */
-    public BulkWriteService intoBulkWriteStream(
-            String database, String table, Schema schema, long timeoutMs, CallOption... options) {
-        return intoBulkWriteStream(database, table, schema, timeoutMs, new AsyncPutListener(), options);
-    }
-
     /**
      * Creates a bulk write stream for efficiently writing data to the server.
      *
-     * @param database the name of the target database
      * @param table the name of the target table
      * @param schema the Arrow schema defining the structure of the data to be written
      * @param timeoutMs the timeout in milliseconds for the write operation
-     * @param metadataListener listener for handling server metadata responses during the write operation
      * @param options optional RPC-layer hints to configure the underlying Flight client call
      * @return a BulkStreamWriter instance that manages the data transfer process
      */
-    public BulkWriteService intoBulkWriteStream(
-            String database,
-            String table,
-            Schema schema,
-            long timeoutMs,
-            PutListener metadataListener,
-            CallOption... options) {
-        FlightDescriptor descriptor = FlightDescriptor.path(database, table);
-        return new BulkWriteService(this, schema, descriptor, metadataListener, timeoutMs, options);
+    public BulkWriteService intoBulkWriteStream(String table, Schema schema, long timeoutMs, CallOption... options) {
+        FlightDescriptor descriptor = FlightDescriptor.path(table);
+        return new BulkWriteService(this, this.allocator, schema, descriptor, timeoutMs, options);
     }
 
     VectorSchemaRoot createSchemaRoot(Schema schema) {
         return VectorSchemaRoot.create(schema, this.allocator);
     }
 
-    ClientStreamListener startPut(
-            FlightDescriptor descriptor, PutListener metadataListener, Runnable onReadyHandler, CallOption... options) {
-        return this.flightClient.startPut(descriptor, metadataListener, onReadyHandler, options);
+    ClientStreamListener startPut(FlightDescriptor descriptor, PutListener metadataListener, CallOption... options) {
+        return this.flightClient.startPut(descriptor, metadataListener, options);
     }
 
     DictionaryProvider newDefaultDictionaryProvider() {
@@ -196,13 +172,13 @@ static class FlightAllocationListener implements AllocationListener {
 
         @Override
         public void onAllocation(long size) {
-            LOG.debug("onAllocation: {}", size);
+            LOG.trace("onAllocation: {}", size);
             ALLOCATION_BYTES.inc(size);
         }
 
         @Override
         public void onRelease(long size) {
-            LOG.debug("onRelease: {}", size);
+            LOG.trace("onRelease: {}", size);
             ALLOCATION_BYTES.dec(size);
         }