content: package com.alibaba.hologres.client.copy.in;
  
  import com.alibaba.hologres.client.copy.WithCopyResult;
  import com.alibaba.hologres.client.model.Record;
  import com.alibaba.hologres.client.model.TableSchema;
  import org.postgresql.core.BaseConnection;
  import org.postgresql.jdbc.TimestampUtils;
  
  import java.io.Closeable;
  import java.io.IOException;
  import java.io.OutputStream;
  import java.nio.ByteBuffer;
  import java.nio.charset.Charset;
  
  /** Record转成copy流. */
  public abstract class RecordOutputStream implements Closeable {
  
      private static final int DEFAULT_MAX_CELL_BUFFER_SIZE = 2 * 1024 * 1024;
  
      protected static final Charset UTF8 = Charset.forName("utf-8");
      protected final TableSchema schema;
  
      private final int maxCellBufferSize;
  
      protected final BaseConnection conn;
      protected final TimestampUtils timestampUtils;
  
      private final OutputStream os;
  
      public RecordOutputStream(
              OutputStream os, TableSchema schema, BaseConnection conn, int maxCellBufferSize) {
          this.schema = schema;
          this.os = os;
          this.maxCellBufferSize = maxCellBufferSize;
          this.conn = conn;
          this.timestampUtils = conn.getTimestampUtils();
      }
  
      boolean closed = false;
      ByteBuffer cellBuffer = ByteBuffer.allocate(10);
      Record currentRecord = null;
      int currentColumnIndex;
  
      public long getResult() {
          if (os instanceof WithCopyResult) {
              return ((WithCopyResult) os).getResult();
          } else {
              return -1;
          }
      }
  
      @Override
      public void close() throws IOException {
          if (!closed) {
              closed = true;
              os.close();
          }
      }
  
      public void putRecord(Record record) throws IOException {
          if (closed) {
              throw new IOException("RecordOutputFormat already closed");
          }
          fillByteBuffer(record);
          writeCellBuffer();
      }
  
      protected void writeCellBuffer() throws IOException {
          cellBuffer.flip();
          os.write(cellBuffer.array(), cellBuffer.position(), cellBuffer.remaining());
          cellBuffer.clear();
      }
  
      protected abstract void fillByteBuffer(Record record) throws IOException;
  
      private void mayIncBuffer(int size) throws IOException {
          if (cellBuffer.remaining() < size) {
              if (cellBuffer.position() + size < maxCellBufferSize) {
                  int target =
                          Math.min(
                                  Math.max(cellBuffer.position() + size, cellBuffer.position() * 2),
                                  maxCellBufferSize);
                  ByteBuffer temp = ByteBuffer.allocate(target);
                  cellBuffer.flip();
                  temp.put(cellBuffer);
                  cellBuffer.clear();
                  cellBuffer = temp;
              } else {
                  throw new IOException(
                          "RecordInputStream cellBuffer exceed max cell size "
                                  + maxCellBufferSize
                                  + " for column "
                                  + schema.getColumn(currentColumnIndex).getName());
              }
          }
      }
  
      /**
       * write byte.
       *
       * @param r
       * @throws IOException
       */
      protected void write(int r) throws IOException {
          mayIncBuffer(1);
          cellBuffer.put((byte) (r & 0xFF));
      }
  
      protected void writeShort(short r) throws IOException {
          mayIncBuffer(2);
          cellBuffer.putShort(r);
      }
  
      protected void writeInt(int r) throws IOException {
          mayIncBuffer(4);
          cellBuffer.putInt(r);
      }
  
      protected void writeFloat(float r) throws IOException {
          mayIncBuffer(4);
          cellBuffer.putFloat(r);
      }
  
      protected void writeDouble(double r) throws IOException {
          mayIncBuffer(8);
          cellBuffer.putDouble(r);
      }
  
      protected void writeLong(long r) throws IOException {
          mayIncBuffer(8);
          cellBuffer.putLong(r);
      }
  
      protected void write(byte[] bytes) throws IOException {
          mayIncBuffer(bytes.length);
          cellBuffer.put(bytes);
      }
  }
  