Implement BeginInsert/InsertData/EndInsert flow

Replace the old (and commented-out) code with the new `BeginInsert` /
`InsertData` / `EndInsert` pattern in `clickhouse-cpp`. This greatly
simplifies the code, as the most of the necessary logic for creating the
insert Block object and tracking the state of the insert is now handled
by the Client object.

Remove the old logic for unwrapping `LowCardinality` columns and fix
`column_append` to properly append to those columns. This preserves
traffic over the wire, I'm told. Thanks to @slabko for working out how
to make that work again.

Copy the test from the old FDW, but comment out the bit working with
arrays, as it has not yet been updated.

Author: theory

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "clickhouse-cpp"]
 	path = vendor/clickhouse-cpp
-	url = https://github.com/ClickHouse/clickhouse-cpp.git
+	url = https://github.com/theory/clickhouse-cpp.git

src/binary.cpp

@@ -285,98 +285,70 @@ static Oid get_corr_postgres_type(const TypeRef & type)
 void ch_binary_insert_state_free(void * c)
 {
 	auto * state = (ch_binary_insert_state *)c;
-	if (state->columns)
+	if (state->insert_block)
 	{
-		/* try to send empty block that sets proper ClickHouse state */
-		if (!state->success)
+		/* Finish the insert to set the proper ClickHouse state */
+		Client * client = (Client *)state->conn->client;
+		try
 		{
-			try
-			{
-				Client * client = (Client *)state->conn->client;
-				client->Insert(state->table_name, Block());
-			}
-			catch (const std::exception & e)
-			{
-				// just ignore, next query will fail
-				elog(NOTICE, "clickhouse_fdw: could not send empty packet");
-			}
+			client->EndInsert();
 		}
-
-		delete (std::vector<clickhouse::ColumnRef> *)state->columns;
+		catch (const std::exception & e)
+		{
+			// just ignore, next query will fail
+			elog(NOTICE, "clickhouse_fdw: could not finish INSERT: - %s", e.what());
+		}
+		delete (Block *)state->insert_block;
 	}
 }
 
 void ch_binary_prepare_insert(void * conn, char * query, ch_binary_insert_state * state)
 {
-	throw std::runtime_error("clickhouse_fdw: XXX ch_binary_prepare_insert not implemented");
-
-// 	std::vector<clickhouse::ColumnRef> * vec = nullptr;
-// 	Client * client = (Client *)((ch_binary_connection_t *)conn)->client;
-
-// 	try
-// 	{
-// 		client->PrepareInsert(
-// 			std::string(query) + " VALUES", [&state, &vec](const Block & sample_block) {
-// 				if (sample_block.GetColumnCount() == 0)
-// 					return true;
-
-// 				vec = new std::vector<clickhouse::ColumnRef>();
-
-// 				state->len = sample_block.GetColumnCount();
-
-// #if PG_VERSION_NUM < 120000
-// 				state->outdesc = CreateTemplateTupleDesc(state->len, false);
-// #else
-// 			    state->outdesc = CreateTemplateTupleDesc(state->len);
-// #endif
-
-// 				for (size_t i = 0; i < state->len; i++)
-// 				{
-// 					bool error = false;
-// 					clickhouse::ColumnRef col = sample_block[i];
-
-// 					auto chtype = col->Type();
-// 					if (chtype->GetCode() == Type::LowCardinality)
-// 					{
-// 						chtype = col->As<ColumnLowCardinality>()->GetNestedType();
-// 					}
-
-// 					Oid pg_type = get_corr_postgres_type(chtype);
-
-// 					vec->push_back(clickhouse::CreateColumnByType(col->Type()->GetName()));
-// 					const char * colname = sample_block.GetColumnName(i).c_str();
-
-// 					/* we can't afford long jumps outside of this function */
-// 					PG_TRY();
-// 					{
-// 						TupleDescInitEntry(
-// 							state->outdesc, (AttrNumber)i + 1, colname, pg_type, -1, 0);
-// 					}
-// 					PG_CATCH();
-// 					{
-// 						error = true;
-// 					}
-// 					PG_END_TRY();
-
-// 					if (error)
-// 						throw std::runtime_error("could not init tuple descriptor");
-// 				}
-
-// 				return true;
-// 			});
-// 	}
-// 	catch (const std::exception & e)
-// 	{
-//         client->ResetConnection();
-
-// 		if (vec != nullptr)
-// 			delete vec;
-
-// 		elog(ERROR, "clickhouse_fdw: error while insert preparation - %s", e.what());
-// 	}
-
-// 	if (vec != nullptr)
-// 		state->columns = (void *)vec;
+	// Start the INSERT.
+	Block block;
+	Client * client = (Client *)((ch_binary_connection_t *)conn)->client;
+	try
+	{
+		block = client->BeginInsert(std::string(query) + " VALUES");
+	}
+	catch (const std::exception & e)
+	{
+		elog(ERROR, "clickhouse_fdw: could prepare insert - %s", e.what());
+	}
+
+	// Setup the column config (or return if no columns).
+	state->len = block.GetColumnCount();
+	if (state->len == 0) return;
+	state->outdesc = CreateTemplateTupleDesc(state->len);
+
+	// Iterate over the list of columns returned by ClickHouse.
+	auto newBlock = new Block();
+	AttrNumber i = 0;
+	for (Block::Iterator bi(block); bi.IsValid(); bi.Next())
+	{
+		// Copy the ClickHouse Column.
+		auto chtype = bi.Type()->GetName();
+		newBlock->AppendColumn(bi.Name(), clickhouse::CreateColumnByType(chtype));
+
+		// Determine the Postgres column type.
+		Oid pg_type = get_corr_postgres_type(bi.Type());
+		const char * colname = bi.Name().c_str();
+
+		PG_TRY();
+		{
+			TupleDescInitEntry(state->outdesc, ++i, colname, pg_type, -1, 0);
+		}
+		PG_CATCH();
+		{
+			// Clean up and re-thrrow.
+			client->ResetConnection();
+			delete newBlock;
+			PG_RE_THROW();
+		}
+		PG_END_TRY();
+	}
+
+	state->insert_block = (ch_insert_block_h *) newBlock;
 }
 
 static void column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, bool isnull)
@@ -504,14 +476,7 @@ static void column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, boo
 					col->As<ColumnEnum16>()->Append(s);
 					break;
 				case Type::Code::LowCardinality: {
-					// XXX Figure out proper value to create and pass to
-					// Append.
-					throw std::runtime_error(
-						"clickhouse_fdw: XXX unsupported column type "
-						+ col->Type()->GetName()
-					);
-					// auto item = ItemView{Type::String, std::string_view(s)};
-					// col->As<ColumnLowCardinality>()->Append(item);
+					col->AsStrict<ColumnLowCardinalityT<ColumnString>>()->Append(s);
 					break;
 				}
 				default:
@@ -573,7 +538,7 @@ static void column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, boo
 			{
 				case Type::Array: {
 					// XXX Figure out proper value to create and pass to
-					// Append.
+					// Append. Then uncomment the test in binary_inserts.sql.
 					throw std::runtime_error(
 						"clickhouse_fdw: XXX unsupported column type "
 						+ col->Type()->GetName()
@@ -607,8 +572,8 @@ void ch_binary_column_append_data(ch_binary_insert_state * state, size_t colidx)
 {
 	try
 	{
-		auto columns = *(std::vector<clickhouse::ColumnRef> *)state->columns;
-		auto col = columns[colidx];
+		auto block = (Block *)state->insert_block;
+		auto col = (*block)[colidx];
 
 		Datum val = state->values[colidx];
 		Oid valtype = TupleDescAttr(state->outdesc, colidx)->atttypid;
@@ -626,16 +591,11 @@ void ch_binary_insert_columns(ch_binary_insert_state * state)
 {
 	try
 	{
-		Block block;
-		auto columns = *(std::vector<clickhouse::ColumnRef> *)state->columns;
-		for (int i = 0; i < state->outdesc->natts; ++i)
-		{
-			Form_pg_attribute att = TupleDescAttr(state->outdesc, i);
-			block.AppendColumn(NameStr(att->attname), columns[i]);
-		}
-
 		Client * client = (Client *)state->conn->client;
-		client->Insert(state->table_name, block);
+		auto block = (Block *)state->insert_block;
+		block->RefreshRowCount();
+		client->InsertData(*block);
+		block->Clear();
 	}
 	catch (const std::exception & e)
 	{

src/include/binary.hh

@@ -6,6 +6,7 @@ extern "C" {
 #endif
 
 typedef struct ch_binary_connection_t ch_binary_connection_t;
+typedef struct ch_insert_block_h ch_insert_block_h;
 typedef struct ch_binary_response_t
 {
 	void			   *values;
@@ -48,7 +49,7 @@ typedef struct {
 	MemoryContextCallback callback;
 
 	TupleDesc	outdesc;
-	void   *columns;	/* std::vector */
+	ch_insert_block_h *insert_block; /* clickhouse::Block */
 	size_t	len;
 	void  *conversion_states;
 	char *table_name;

src/pglink.c

@@ -729,7 +729,6 @@ binary_insert_tuple(void *istate, TupleTableSlot *slot)
 	else
 	{
 		ch_binary_insert_columns(state);
-		state->success = true;
 	}
 }