allow executing COPY INTO in a cluster

[flaneur2020]

Summary

i'm running a COPY INTO in my cluster with a 8 replicas, but it seems only utilized one replica to execute the COPY INTO statement:

it'd be 8x faster if the COPY INTO statement could utilize the other instances in the cluster.

[sundy-li]

Is it a single file ?

[PsiACE]

Is it a single file ?

One hundred files, obtained by cutting the ontime dataset.

[sundy-li]

Ok, seems parallel copy only works in single query mode.

https://github.com/datafuselabs/databend/blob/f152cbe7edc96fe5850982ef40d9c04c57ecc94e/query/src/interpreters/interpreter_copy.rs

     if ctx.get_settings().get_enable_new_processor_framework()? != 0
            && self.ctx.get_cluster().is_empty()
        {
            table.append2(ctx.clone(), &mut pipeline)?;
            pipeline.set_max_threads(settings.get_max_threads()? as usize);

            let async_runtime = ctx.get_storage_runtime();
            let query_need_abort = ctx.query_need_abort();
            let executor =
                PipelineCompleteExecutor::try_create(async_runtime, query_need_abort, pipeline)?;

            executor.execute()?;
            return Ok(ctx.consume_precommit_blocks());
        }

[zhang2014]

Distributed copy into need #6253(exchange precommit block in cluster nodes).

[sundy-li]

@RinChanNOWWW you can try this issue, it's ready to do now.

[BohuTANG]

I think we can make the <internal/external-stage, remote location> as a special storage engine, then we can get the file list as table source, and optimize the files to the distribution cluster. Also, this will be the basement for:
#7228 and #7211

I would ping @dantengsky, he is doing a similar storage engine (pre-sign), if some codes need refactoring, please let us know:)

[RinChanNOWWW]

I think we can make the <internal/external-stage, remote location> as a special storage engine, then we can get the file list as table source

Then we can convert copy into t from @stage into insert into t select from @stage and achieve distributed copy into by #7501.

[BohuTANG]

@RinChanNOWWW

Please take a look #7502
We are going to make the catalog to meet these requirements, work is in progress by @dantengsky
If you are interested, you can ping and talk with dantengsky :)

[PsiACE]

#11943