Limit streaming scans to the binary driver

The PR originally carried coroutine-based HTTP streaming and the\nvendored minicoro dependency even after the binary path moved to\nthe new clickhouse-cpp pull API. That made the diff much larger\nthan the final design and left the PR description out of date.\n\nDrop the unused HTTP streaming implementation, remove minicoro\nfrom the build and submodules, and restore the buffered HTTP\npath. Keep only the binary scan changes: the FDW streaming hook,\nthe block-backed binary row reader, and the fetch_size plumbing.\nAlso restore the buffered binary cancel callback and remove the\nnew C-linkage warning from binary_internal.hh.

Author: iskakaushik

.gitmodules

@@ -1,6 +1,3 @@
 [submodule "clickhouse-cpp"]
 	path = vendor/clickhouse-cpp
 	url = git@github.com:iskakaushik/clickhouse-cpp.git
-[submodule "vendor/minicoro"]
-	path = vendor/minicoro
-	url = https://github.com/edubart/minicoro.git

Makefile

@@ -33,9 +33,8 @@ ifndef CH_BUILD
 endif
 
 # clickhouse-cpp source and build directories.
-VENDOR_DIR = vendor
-CH_CPP_DIR = $(VENDOR_DIR)/clickhouse-cpp
-CH_CPP_BUILD_DIR = $(VENDOR_DIR)/_build/$(OS)-$(ARCH)-$(CH_BUILD)-$(shell git submodule status $(CH_CPP_DIR) | awk '{print substr($$1, 0, 7)}')
+CH_CPP_DIR = vendor/clickhouse-cpp
+CH_CPP_BUILD_DIR = vendor/_build/$(OS)-$(ARCH)-$(CH_BUILD)-$(shell git submodule status $(CH_CPP_DIR) | awk '{print substr($$1, 0, 7)}')
 
 # List the clickhouse-cpp libraries we require.
 CH_CPP_LIB = $(CH_CPP_BUILD_DIR)/clickhouse/libclickhouse-cpp-lib$(DLSUFFIX)
@@ -57,7 +56,7 @@ else
 endif
 
 # Add include directories.
-PG_CPPFLAGS = -I./src/include -I$(CH_CPP_DIR) -I$(CH_CPP_DIR)/contrib/absl -I$(VENDOR_DIR)/minicoro
+PG_CPPFLAGS = -I./src/include -I$(CH_CPP_DIR) -I$(CH_CPP_DIR)/contrib/absl
 
 # Include other libraries compiled into clickhouse-cpp.
 PG_LDFLAGS = -lstdc++ -lssl -lcrypto $(shell $(CURL_CONFIG) --libs)
@@ -76,7 +75,7 @@ endif
 # Clean up the clickhouse-cpp build directory and generated files.
 EXTRA_CLEAN = sql/$(EXTENSION)--$(EXTVERSION).sql src/fdw.c compile_commands.json test/schedule $(EXTENSION)-$(DISTVERSION).zip
 ifndef NO_VENDOR_CLEAN
-	EXTRA_CLEAN += $(VENDOR_DIR)
+	EXTRA_CLEAN += $(CH_CPP_BUILD_DIR)
 endif
 
 # Import PGXS.
@@ -102,12 +101,8 @@ $(shlib): $(CH_CPP_LIB) $(OBJS)
 $(CH_CPP_DIR)/CMakeLists.txt:
 	git submodule update --init
 
-# Clone minicoro submodule.
-$(VENDOR_DIR)/minicoro/minicoro.h:
-	git submodule update --init
-
-# Require the vendored libraries.
-$(OBJS): $(CH_CPP_DIR)/CMakeLists.txt $(VENDOR_DIR)/minicoro/minicoro.h
+# Require the vendored clickhouse-cpp.
+$(OBJS): $(CH_CPP_DIR)/CMakeLists.txt
 
 # Build clickhouse-cpp.
 $(CH_CPP_LIB): export CXXFLAGS=-fPIC
@@ -117,7 +112,7 @@ $(CH_CPP_LIB): $(CH_CPP_DIR)/CMakeLists.txt # Sync with "Reset Vendor Timestamp"
 	cmake --build $(CH_CPP_BUILD_DIR) --parallel $$(nproc) --target all
 
 # Require the versioned C source and SQL script.
-all: sql/$(EXTENSION)--$(EXTVERSION).sql src/fdw.c $(VENDOR_DIR)/minicoro/minicoro.h
+all: sql/$(EXTENSION)--$(EXTVERSION).sql src/fdw.c
 
 # Versioned SQL script.
 sql/$(EXTENSION)--$(EXTVERSION).sql: sql/$(EXTENSION).sql

src/binary.cpp

@@ -233,6 +233,12 @@ extern "C"
 			client->Select(clickhouse::Query(query->sql)
 						   .SetQuerySettings(ch_binary_settings(query))
 						   .SetParams(ch_binary_params(query))
+						   .OnProgress(
+						   [&check_cancel](const Progress &)
+						   {
+							   if (check_cancel && check_cancel())
+								   throw std::runtime_error("query was canceled");
+						   })
 						   .OnDataCancelable(
 						   [&resp, &values, &check_cancel](const Block &block)
 						   {

src/binary_streaming.cpp

@@ -35,50 +35,26 @@ struct ch_binary_streaming_state
 	std::unique_ptr<bool[]> nulls;
 	size_t		columns_count = 0;
 
-	const char *error = nullptr;
-	char	   *owned_error = nullptr;
+	std::optional<std::string> error;
 	bool		(*check_cancel) (void) = nullptr;
 
 	Client	   *client = nullptr;
 	std::string	sql;
 	QuerySettings settings;
 	QueryParams	params;
 
-	~ch_binary_streaming_state()
-	{
-		free(owned_error);
-	}
-
 	void
-	SetBorrowedError(const char *message)
+	SetError(const char *message)
 	{
 		if (error)
 			return;
-		error = message;
+		error.emplace(message ? message : kBinaryStreamingOom);
 	}
 
-	void
-	SetOwnedError(const char *message)
+	const char *
+	GetError() const
 	{
-		char	   *copy;
-
-		if (error)
-			return;
-		if (!message)
-		{
-			SetBorrowedError(kBinaryStreamingOom);
-			return;
-		}
-
-		copy = strdup(message);
-		if (!copy)
-		{
-			SetBorrowedError(kBinaryStreamingOom);
-			return;
-		}
-
-		owned_error = copy;
-		error = owned_error;
+		return error ? error->c_str() : nullptr;
 	}
 };
 
@@ -93,7 +69,7 @@ binary_streaming_fill_block(ch_binary_streaming_state * st)
 		{
 			if (st->check_cancel && st->check_cancel())
 			{
-				st->SetBorrowedError(kBinaryStreamingCanceled);
+				st->SetError(kBinaryStreamingCanceled);
 				st->done = true;
 				return false;
 			}
@@ -102,7 +78,7 @@ binary_streaming_fill_block(ch_binary_streaming_state * st)
 		}
 		catch (const std::exception & e)
 		{
-			st->SetOwnedError(e.what());
+			st->SetError(e.what());
 			st->done = true;
 			return false;
 		}
@@ -115,7 +91,7 @@ binary_streaming_fill_block(ch_binary_streaming_state * st)
 			}
 			catch (const std::exception & e)
 			{
-				st->SetOwnedError(e.what());
+				st->SetError(e.what());
 			}
 			st->current_block.reset();
 			st->have_block = false;
@@ -130,7 +106,7 @@ binary_streaming_fill_block(ch_binary_streaming_state * st)
 		if (st->columns_count != 0 &&
 			block->GetColumnCount() != st->columns_count)
 		{
-			st->SetBorrowedError("columns mismatch in blocks");
+			st->SetError("columns mismatch in blocks");
 			st->done = true;
 			return false;
 		}
@@ -171,7 +147,7 @@ extern "C"
 		}
 		catch (const std::exception & e)
 		{
-			st->SetOwnedError(e.what());
+			st->SetError(e.what());
 			st->done = true;
 			return st;
 		}
@@ -216,7 +192,7 @@ extern "C"
 			st->nulls.reset(new (std::nothrow) bool[st->columns_count]);
 			if (!st->coltypes || !st->values || !st->nulls)
 			{
-				st->SetBorrowedError(kBinaryStreamingOom);
+				st->SetError(kBinaryStreamingOom);
 				return false;
 			}
 		}
@@ -231,7 +207,7 @@ extern "C"
 		}
 		catch (const std::exception & e)
 		{
-			st->SetOwnedError(e.what());
+			st->SetError(e.what());
 			return false;
 		}
 
@@ -264,7 +240,7 @@ extern "C"
 	const char *
 	ch_binary_streaming_error(ch_binary_streaming_state * st)
 	{
-		return st ? st->error : NULL;
+		return st ? st->GetError() : NULL;
 	}
 
 	void

src/fdw.c.in

@@ -956,6 +956,9 @@ clickhouseIterateForeignScan(ForeignScanState * node)
 	TupleDesc	tupdesc;
 	ch_query	query = new_query(fsstate->query, fsstate->numParams, fsstate->param_values);
 
+	/* Allow query cancel (e.g. Ctrl+C) between tuple fetches. */
+	CHECK_FOR_INTERRUPTS();
+
 	/* make query if needed */
 	if (fsstate->ch_cursor == NULL)
 	{