bug: distributed replace into with attachment not work as expected

[dantengsky]

Summary

distributed replace into with attachment not work as expected, with error "code":1001,"message":"Invalid argument error: The argument fields must be consistent with the columns' schema. Use e.g. &arrow2::io::flight::default_ipc_fields(&schema.fields)","detail":"".

(java)client side error:

Exception in thread "main" java.sql.SQLException: Error executing query: SQL: replace into test_prepare_statement on(a) values (?,?) Query failed: Optional[{"id":"64927fde-69ca-4934-8cec-47b973136904","session_id":"c0eebd60-bb2f-47bf-a73a-bf229e0ae31a","node_id":"3tn8KnU6C7T1BlMRP7VFD1","state":"Failed","session":{"database":"default","role":"public","settings":{},"txn_state":"AutoCommit","last_server_info":{"id":"3tn8KnU6C7T1BlMRP7VFD1","start_time":"2024-06-27T17:55:37.133+08:00"},"last_query_ids":["64927fde-69ca-4934-8cec-47b973136904"]},"error":{"code":1001,"message":"Invalid argument error: The argument `fields` must be consistent with the columns' schema. Use e.g. &arrow2::io::flight::default_ipc_fields(&schema.fields)","detail":""},"warnings":[],"has_result_set":true,"schema":[{"name":"a","type":"Nullable(Int32)"},{"name":"b","type":"Nullable(String)"}],"data":[],"affect":null,"stats":{"scan_progress":{"rows":1,"bytes":4},"write_progress":{"rows":0,"bytes":0},"result_progress":{"rows":0,"bytes":0},"total_scan":{"rows":1,"bytes":4},"running_time_ms":446},"stats_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904","final_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/final","next_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/final","kill_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/kill"}] cause: null
	at com.databend.jdbc.DatabendStatement.internalExecute(DatabendStatement.java:215)
	at com.databend.jdbc.DatabendPreparedStatement.executeBatchByAttachment(DatabendPreparedStatement.java:299)
	at com.databend.jdbc.DatabendPreparedStatement.executeBatch(DatabendPreparedStatement.java:384)
	at com.databend.jdbc.examples.Examples.main(Examples.java:64)
Caused by: java.lang.RuntimeException: Query failed: Optional[{"id":"64927fde-69ca-4934-8cec-47b973136904","session_id":"c0eebd60-bb2f-47bf-a73a-bf229e0ae31a","node_id":"3tn8KnU6C7T1BlMRP7VFD1","state":"Failed","session":{"database":"default","role":"public","settings":{},"txn_state":"AutoCommit","last_server_info":{"id":"3tn8KnU6C7T1BlMRP7VFD1","start_time":"2024-06-27T17:55:37.133+08:00"},"last_query_ids":["64927fde-69ca-4934-8cec-47b973136904"]},"error":{"code":1001,"message":"Invalid argument error: The argument `fields` must be consistent with the columns' schema. Use e.g. &arrow2::io::flight::default_ipc_fields(&schema.fields)","detail":""},"warnings":[],"has_result_set":true,"schema":[{"name":"a","type":"Nullable(Int32)"},{"name":"b","type":"Nullable(String)"}],"data":[],"affect":null,"stats":{"scan_progress":{"rows":1,"bytes":4},"write_progress":{"rows":0,"bytes":0},"result_progress":{"rows":0,"bytes":0},"total_scan":{"rows":1,"bytes":4},"running_time_ms":446},"stats_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904","final_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/final","next_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/final","kill_uri":"/v1/query/64927fde-69ca-4934-8cec-47b973136904/kill"}]
	at com.databend.client.DatabendClientV1.executeInternal(DatabendClientV1.java:184)
	at com.databend.client.DatabendClientV1.execute(DatabendClientV1.java:205)
	at com.databend.client.DatabendClientV1.<init>(DatabendClientV1.java:91)
	at com.databend.jdbc.DatabendConnection.startQuery(DatabendConnection.java:604)
	at com.databend.jdbc.DatabendStatement.internalExecute(DatabendStatement.java:188)
	... 3 more

server side error log:

89d963b4-d696-47e6-a993-b6451048ca51 2024-06-27T10:12:45.457986Z  INFO databend_query::servers::http::v1::http_query_handlers: http_query_handlers.rs:384 http query initial response to http query_id=89d963b4-d696-
47e6-a993-b6451048ca51, state=ResponseState { has_result_set: Some(true), schema: [QueryResponseField { name: "a", type: "Nullable(Int32)" }, QueryResponseField { name: "b", type: "Nullable(String)" }], running_ti
me_ms: 194, progresses: Progresses { scan_progress: ProgressValues { rows: 0, bytes: 0 }, write_progress: ProgressValues { rows: 0, bytes: 0 }, result_progress: ProgressValues { rows: 0, bytes: 0 }, total_scan: Pr
ogressValues { rows: 1, bytes: 4 } }, state: Failed, affect: None, error: Some(FromStdError. Code: 1001, Text = Invalid argument error: The argument `fields` must be consistent with the columns' schema. Use e.g. &
arrow2::io::flight::default_ipc_fields(&schema.fields).

   0: databend_common_exception::exception_backtrace::capture
             at src/common/exception/src/exception_backtrace.rs:82:17
   1: databend_common_exception::exception::ErrorCode::from_std_error
             at src/common/exception/src/exception.rs:277:24
   2: databend_common_exception::exception_into::<impl core::convert::From<databend_common_arrow::arrow::error::Error> for databend_common_exception::exception::ErrorCode>::from
             at src/common/exception/src/exception_into.rs:115:18
   3: <core::result::Result<T,F> as core::ops::try_trait::FromResidual<core::result::Result<core::convert::Infallible,E>>>::from_residual
             at /rustc/8ace7ea1f7cbba7b4f031e66c54ca237a0d65de6/library/core/src/result.rs:1959:27
   4: databend_query::servers::flight::v1::exchange::serde::exchange_serializer::serialize_block
             at src/query/service/src/servers/flight/v1/exchange/serde/exchange_serializer.rs:226:13
   5: <databend_query::servers::flight::v1::exchange::serde::exchange_serializer::TransformScatterExchangeSerializer as databend_common_pipeline_transforms::processors::transforms::transform::BlockMetaTransform<da
tabend_query::servers::flight::v1::exchange::exchange_transform_shuffle::ExchangeShuffleMeta>>::transform
             at src/query/service/src/servers/flight/v1/exchange/serde/exchange_serializer.rs:194:26
   6: <databend_common_pipeline_transforms::processors::transforms::transform::BlockMetaTransformer<B,T> as databend_common_pipeline_core::processors::processor::Processor>::process
             at src/query/pipeline/transforms/src/processors/transforms/transform.rs:305:37
   7: <alloc::boxed::Box<T> as databend_common_pipeline_core::processors::processor::Processor>::process
             at src/query/pipeline/core/src/processors/processor.rs:230:9
   8: databend_common_pipeline_core::processors::processor::ProcessorPtr::process
             at src/query/pipeline/core/src/processors/processor.rs:169:9
   9: databend_query::pipelines::executor::executor_worker_context::ExecutorWorkerContext::execute_sync_task
             at src/query/service/src/pipelines/executor/executor_worker_context.rs:169:9
  10: databend_query::pipelines::executor::executor_worker_context::ExecutorWorkerContext::execute_task
             at src/query/service/src/pipelines/executor/executor_worker_context.rs:132:52
  11: databend_query::pipelines::executor::query_pipeline_executor::QueryPipelineExecutor::execute_single_thread
             at src/query/service/src/pipelines/executor/query_pipeline_executor.rs:420:35
  12: databend_query::pipelines::executor::query_pipeline_executor::QueryPipelineExecutor::execute_threads::{{closure}}::{{closure}}
             at src/query/service/src/pipelines/executor/query_pipeline_executor.rs:379:27
  13: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once
             at /rustc/8ace7ea1f7cbba7b4f031e66c54ca237a0d65de6/library/core/src/panic/unwind_safe.rs:272:9
  14: std::panicking::try::do_call
             at /rustc/8ace7ea1f7cbba7b4f031e66c54ca237a0d65de6/library/std/src/panicking.rs:554:40
  15: __rust_try
  16: std::panicking::try
             at /rustc/8ace7ea1f7cbba7b4f031e66c54ca237a0d65de6/library/std/src/panicking.rs:518:19
  17: std::panic::catch_unwind
             at /rustc/8ace7ea1f7cbba7b4f031e66c54ca237a0d65de6/library/std/src/panic.rs:142:14
  18: databend_common_base::runtime::catch_unwind::catch_unwind
...

version: v1.2.541-nightly

how to reproduce:

• cluster of v1.2.541 query nodes
• storage type should support pre-sign(e.g. minio)
• set global enable_distributed_replace_into = 1;
• run the attached java prog, e.g. java -cp .:databend-jdbc-0.2.7.jar com.databend.jdbc.examples.Examples

Examples.java.tar.gz