[CH] New byte buffer takes most of time in SourceFromJavalter::generate

[taiyang-li]

Description

Reproduce sqls:

create table default.test_data_left as select id as id, id % 10 as x, id % 1000 as y, id -1 as z , 'left' as w from range(10000000);

create table default.test_data_right as select id as id, id % 10 as x, id % 1000 as y, id -1 as z , 'right' as w from range(10001000);

select avg(id1), avg(id2) from (select l.id as id1, r.id as id2 from test_data_left as l left join test_data_right as r on l.id = r.id);

Two issues cc @baibaichen

• New byte buffer takes most of time in SourceFromJavalter::generate. It is not as expected
• Wrong use of memory.m_capacity in ReadBufferFromJavaInputStream::readFromJava, should use memory.m_size.

bool ReadBufferFromJavaInputStream::nextImpl()
{
    int count = readFromJava();
    if (count > 0)
        working_buffer.resize(count);
    return count > 0;
}
int ReadBufferFromJavaInputStream::readFromJava() const
{
    GET_JNIENV(env)
    jint count = safeCallIntMethod(
        env, java_in, ShuffleReader::input_stream_read, reinterpret_cast<jlong>(working_buffer.begin()), memory.m_capacity);
    CLEAN_JNIENV
    return count;
}

@Override
  public long read(long destAddress, long maxReadSize) {
    return GlutenException.wrap(
        () -> {
          int maxReadSize32 = Math.toIntExact(maxReadSize);
          if (buffer == null || maxReadSize32 > buffer.length) {
            this.buffer = new byte[maxReadSize32];
          }
          // The code conducts copy as long as 'in' wraps off-heap data,
          // which is about to be moved to heap
          int read = in.read(buffer, 0, maxReadSize32);
          if (read == -1 || read == 0) {
            return 0;
          }
          // The code conducts copy, from heap to off-heap
          // memCopyFromHeap(buffer, destAddress, read);
          PlatformDependent.copyMemory(buffer, 0, destAddress, read);
          bytesRead += read;
          return read;
        });

[zhanglistar]

optoruntime::new_array_c可能是传入的memory.m_capacity过大，另外jdk中会对内存进行memset，导致该函数占用过多的时间。

[taiyang-li]

原因：查询运行过程中，有26200次new byte[1024*1024] 操作，平均每个task有78次，总耗时8s, 而查询耗时也就30+s

问题：为什么会走带copy的OnHeapCopyShuffleInputStream，没走zero-copy的LowCopyNettyShuffleInputStream

调用链

CHColumnarBatchSerializerInstance.deserializeStream
CHStreamReader.CHStreamReader
CHShuffleReadStreamFactory.create

public static ShuffleInputStream create(
      InputStream in, boolean forceCompress, boolean isCustomizedShuffleCodec) {
    final InputStream unwrapped = unwrapInputStream(in, forceCompress, isCustomizedShuffleCodec);
    if (unwrapped != null) {
      return createCompressedShuffleInputStream(in, unwrapped);
    }
    return new OnHeapCopyShuffleInputStream(in, false);
  }

  private static InputStream unwrapInputStream(
      InputStream in, boolean forceCompress, boolean isCustomizedShuffleCodec) {
    if (forceCompress) {
      return unwrapSparkInputStream(in);
    } else if (isCustomizedShuffleCodec) {
      return unwrapSparkWithCompressedInputStream(in);
    }
    return null;
  }

由于我的local环境中并未设置celeborn作为shuffle manager, 因此最终走了OnHeapCopyShuffleInputStream。而OnHeapCopyShuffleInputStream目前的实现还不是很高效，最终导致了标题中描述的问题。

[taiyang-li]

修改配置后 " --conf spark.shuffle.manager=org.apache.spark.shuffle.gluten.celeborn.CelebornShuffleManager"
火焰图如下：

[zzcclp]

原因：查询运行过程中，有26200次new byte[1024*1024] 操作，平均每个task有78次，总耗时8s, 而查询耗时也就30+s

问题：为什么会走带copy的OnHeapCopyShuffleInputStream，没走zero-copy的LowCopyNettyShuffleInputStream

调用链

CHColumnarBatchSerializerInstance.deserializeStream
CHStreamReader.CHStreamReader
CHShuffleReadStreamFactory.create

public static ShuffleInputStream create(
      InputStream in, boolean forceCompress, boolean isCustomizedShuffleCodec) {
    final InputStream unwrapped = unwrapInputStream(in, forceCompress, isCustomizedShuffleCodec);
    if (unwrapped != null) {
      return createCompressedShuffleInputStream(in, unwrapped);
    }
    return new OnHeapCopyShuffleInputStream(in, false);
  }

  private static InputStream unwrapInputStream(
      InputStream in, boolean forceCompress, boolean isCustomizedShuffleCodec) {
    if (forceCompress) {
      return unwrapSparkInputStream(in);
    } else if (isCustomizedShuffleCodec) {
      return unwrapSparkWithCompressedInputStream(in);
    }
    return null;
  }

由于我的local环境中并未设置celeborn作为shuffle manager, 因此最终走了OnHeapCopyShuffleInputStream。而OnHeapCopyShuffleInputStream目前的实现还不是很高效，最终导致了标题中描述的问题。

这里可能要看下你本地调用连，理应要走 LowCopyFileSegmentShuffleInputStream 这个，因为是从本地文件直接读取，按理走这里。