Method for Ingesting Arrays of Timestamps for Array(Datetime64(3)) Clickhouse columns

[alexshanabrook]

Hi! I'm trying to use the clickhouse connector on data that includes an array of timestamps that should be stored in a column of type Array(Datetime64(3)), but I'm getting an error from the connector with the following stack trace:

com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:64).. 11 more
Caused by: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 33. DB::Exception: Cannot read all data. Bytes read: 0. Bytes expected: 1.: (at row 1):
 While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 23.1.3.5 (official build)), server ClickHouseNode @-38134688
java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertRawBinary(ClickHouseWriter.java:512)

Can anyone help me figure out why I'm getting this error? The reason I think it has to do with the array of timestamps is that I've gotten a connector using the same config to successfully ingest data without an array of timestamps. This line of the connector source code implies the issue is coming from Clickhouse—the rest of the stack trace is swallowed, so I can't provide it—but I suspect the issue still has to do with how I'm setting up the connector for the data, since we've also been able to use other connectors for arrays of timestamps successfully in the past.

Update: I found the Clickhouse stack trace for the same error below:

2023.11.14 19:13:56.667570 [ 223797 ] {7232b565-1ea0-477a-8bfe-79e74f6c58b9} <Error> executeQuery: Code: 33. DB::Exception: Cannot read all data. Bytes read: 0. Bytes expected: 1.: (at row 1)
: While executing BinaryRowInputFormat. (CANNOT_READ_ALL_DATA) (version 23.1.3.5 (official build)) (from 10.22.34.71:49822) (in query: INSERT INTO `inference_ed1b613d_c495_4b92_9893_cbd6539b10ca` FORMAT RowBinary ), Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0xddb0df5 in /usr/bin/clickhouse
1. ? @ 0x8a2caec in /usr/bin/clickhouse
2. DB::SerializationNullable::deserializeBinary(DB::IColumn&, DB::ReadBuffer&, DB::FormatSettings const&) const @ 0x12789645 in /usr/bin/clickhouse
3. DB::SerializationArray::deserializeBinary(DB::IColumn&, DB::ReadBuffer&, DB::FormatSettings const&) const @ 0x127675d3 in /usr/bin/clickhouse
4. DB::BinaryFormatReader::readField(DB::IColumn&, std::__1::shared_ptr<DB::IDataType const> const&, std::__1::shared_ptr<DB::ISerialization const> const&, bool, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>> const&) @ 0x1475a9b6 in /usr/bin/clickhouse
5. DB::RowInputFormatWithNamesAndTypes::readRow(std::__1::vector<COW<DB::IColumn>::mutable_ptr<DB::IColumn>, std::__1::allocator<COW<DB::IColumn>::mutable_ptr<DB::IColumn>>>&, DB::RowReadExtension&) @ 0x1475da38 in /usr/bin/clickhouse
6. DB::IRowInputFormat::generate() @ 0x1473cf0e in /usr/bin/clickhouse
7. DB::ISource::tryGenerate() @ 0x146c1575 in /usr/bin/clickhouse
8. DB::ISource::work() @ 0x146c10e6 in /usr/bin/clickhouse
9. DB::ExecutionThreadContext::executeTask() @ 0x146db366 in /usr/bin/clickhouse
10. DB::PipelineExecutor::executeStepImpl(unsigned long, std::__1::atomic<bool>*) @ 0x146d047c in /usr/bin/clickhouse
11. DB::PipelineExecutor::executeImpl(unsigned long) @ 0x146ceeb9 in /usr/bin/clickhouse
12. DB::PipelineExecutor::execute(unsigned long) @ 0x146cec00 in /usr/bin/clickhouse
13. DB::CompletedPipelineExecutor::execute() @ 0x146cd383 in /usr/bin/clickhouse
14. DB::executeQuery(DB::ReadBuffer&, DB::WriteBuffer&, bool, std::__1::shared_ptr<DB::Context>, std::__1::function<void (std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>> const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>> const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>> const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char>> const&)>, std::__1::optional<DB::FormatSettings> const&) @ 0x13811bd5 in /usr/bin/clickhouse
15. DB::HTTPHandler::processQuery(DB::HTTPServerRequest&, DB::HTMLForm&, DB::HTTPServerResponse&, DB::HTTPHandler::Output&, std::__1::optional<DB::CurrentThread::QueryScope>&) @ 0x146155f0 in /usr/bin/clickhouse
16. DB::HTTPHandler::handleRequest(DB::HTTPServerRequest&, DB::HTTPServerResponse&) @ 0x1461960e in /usr/bin/clickhouse
17. DB::HTTPServerConnection::run() @ 0x14694e9d in /usr/bin/clickhouse
18. Poco::Net::TCPServerConnection::start() @ 0x1753e2f4 in /usr/bin/clickhouse
19. Poco::Net::TCPServerDispatcher::run() @ 0x1753fc9b in /usr/bin/clickhouse
20. Poco::PooledThread::run() @ 0x176cf0a7 in /usr/bin/clickhouse
21. Poco::ThreadImpl::runnableEntry(void*) @ 0x176ccadd in /usr/bin/clickhouse
22. ? @ 0x7f2684ad9609 in ?
23. clone @ 0x7f26849fe133 in ?

[Paultagoras]

Hi! Could you run "SHOW CREATE TABLE" on that and share here? Also, do you happen to have any sample data as an example?

[alexshanabrook]

Yes, thank you!
The DDL for the Clickhouse table is:

CREATE TABLE default.inference_ed1b613d_c495_4b92_9893_cbd6539b10ca
(

    `inference_id` UUID,

    `partner_inference_id` String,

    `received_timestamp` DateTime64(3),

    `inference_timestamp` DateTime64(3),

    `batch_id` LowCardinality(String),

    `bill_pay_history_timestamps` Array(DateTime64(3)),

    `bill_pay_history_values` Array(Nullable(Float64)),

    `predicted_recommendations_item_id` Array(String),

    `predicted_recommendations_item_label` Array(String),

    `predicted_recommendations_item_score` Array(Float64)
)
ENGINE = ReplicatedMergeTree('/clickhouse/{installation}/{cluster}/tables/{shard}/default/inference_ed1b613d_c495_4b92_9893_cbd6539b10ca',
 '{replica}')
PARTITION BY toYYYYMM(inference_timestamp)
ORDER BY (batch_id,
 inference_timestamp,
 inference_id)
SETTINGS index_granularity = 8192;

Here's an example of sample data (any timestamps are represented as strings in the example but they're sent to the connector as time.Time objects, not strings):

  {
      "batch_id": "test-batch",
      "partner_inference_id": "1",
      "inference_id": "some_uuid",
      "inference_timestamp": "2020-07-26T23:00:13.656449Z",
      "bill_pay_history_timestamps": ["2023-10-04T00:00:00Z", "2023-10-04T00:00:00Z"],
      "bill_pay_history_values": [200.0, 500.0],
      "predicted_recommendations_label": ["banana", "apple"],
      "predicted_recommendations_item_id": ["item1", "item2"],
      "predicted_recommendations_score": [0.324, 0.024]
  }

The bill_pay_history_timestamps column is the one I think might be causing the issue

Update: I actually think the bill_pay_history_values column is the one causing the issue after seeing the calls to deserializeArray and deserializeNullable in the clickhouse stack trace since that column holds arrays of nullable doubles—still not sure why this could be happening though.

[alexshanabrook]

Update: (cc @Paultagoras ) after looking further into the code I think I found the bug causing this behavior, so I opened an issue here instead & I'll close out this discussion