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.

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,60 @@ 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;
+	Block *block = nullptr;
+	Client * client = (Client *)((ch_binary_connection_t *)conn)->client;
+	try
+	{
+		// The Client retursn a `unique_ptr`. Grab the internal pointer to
+		// take ownership of it and manage its memory ourselves.
+		block = client->BeginInsert(std::string(query) + " VALUES").release();
+	}
+	catch (const std::exception & e)
+	{
+		elog(ERROR, "clickhouse_fdw: could prepare insert - %s", e.what());
+	}
+	state->len = block->GetColumnCount();
+	if (state->len == 0) return;
+	state->outdesc = CreateTemplateTupleDesc(state->len);
+
+	AttrNumber i = 0;
+	for (Block::Iterator bi(*block); bi.IsValid(); bi.Next())
+	{
+		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();
+		{
+			client->ResetConnection();
+			delete block;
+		}
+		PG_END_TRY();
+	}
+
+	state->insert_block = (ch_insert_block_h *) block;
 }
 
 static void column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, bool isnull)
@@ -573,7 +535,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 +569,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 +588,9 @@ 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;
+		client->InsertData(*block);
 	}
 	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;
 	}
 }