Implement binary insert with PreparedInsert

Replace the old (and commented-out) code with the new `PreparedInsert`
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 `PreparedInsert`
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

@@ -47,6 +47,8 @@ using namespace clickhouse;
 #define HOST_TO_BIG_ENDIAN_64(x) htobe64(x)
 #endif
 
+#define GET_INSERTER(state) (Client::PreparedInsert *)state->inserter
+
 /* palloc which will throw exceptions */
 static void * exc_palloc(Size size)
 {
@@ -285,98 +287,54 @@ 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->inserter)
 	{
-		/* try to send empty block that sets proper ClickHouse state */
+		/* Finish the insert to set the proper ClickHouse state */
 		if (!state->success)
 		{
+			auto inserter = GET_INSERTER(state);
 			try
 			{
-				Client * client = (Client *)state->conn->client;
-				client->Insert(state->table_name, Block());
+				inserter->Finish();
 			}
 			catch (const std::exception & e)
 			{
 				// just ignore, next query will fail
-				elog(NOTICE, "clickhouse_fdw: could not send empty packet");
+				elog(NOTICE, "clickhouse_fdw: could not finish INSERT: - %s", e.what());
 			}
 		}
-
-		delete (std::vector<clickhouse::ColumnRef> *)state->columns;
+		delete (Client::PreparedInsert *)state->inserter;
 	}
 }
 
 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;
+	Client * client = (Client *)((ch_binary_connection_t *)conn)->client;
+	auto inserter = client->PrepareInsert(std::string(query) + " VALUES");
+	auto block = inserter->GetBlock();
+	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 inserter;
+		}
+		PG_END_TRY();
+	}
+
+	state->inserter = (ch_prepared_insert_h *) inserter;
 }
 
 static void column_append(clickhouse::ColumnRef col, Datum val, Oid valtype, bool isnull)
@@ -573,7 +531,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 +565,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 inserter = GET_INSERTER(state);
+		auto col = (*inserter->GetBlock())[colidx];
 
 		Datum val = state->values[colidx];
 		Oid valtype = TupleDescAttr(state->outdesc, colidx)->atttypid;
@@ -626,16 +584,8 @@ 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 inserter = GET_INSERTER(state);
+		inserter->Finish();
 	}
 	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_prepared_insert_h ch_prepared_insert_h;
 typedef struct ch_binary_response_t
 {
 	void			   *values;
@@ -48,7 +49,7 @@ typedef struct {
 	MemoryContextCallback callback;
 
 	TupleDesc	outdesc;
-	void   *columns;	/* std::vector */
+	ch_prepared_insert_h *inserter; /* Client::PreparedInsert */
 	size_t	len;
 	void  *conversion_states;
 	char *table_name;