Streaming values while also pre-checking if they already exist.

[Dich0tomy]

Hello, I have a code that looks something like this:

auto work = pqxx::work(connection);

auto stream = pqxx::stream_to::table(
	work,
	"table",
	std::vector<std::string> {
		"foo",
		"bar",
		"baz",
	}
);

for(auto const& d : some_data) {
	if(data_already_in_database(work, d.bar)) {
		// return error
	}

	stream.write_values(
		c.foo,
		c.bar,
		c.baz
	);
}

stream.complete();

work.commit();

// return some success

data_already_in_database is defined +- like this:

auto data_already_in_database(pqxx::work& work, std::string_view bar)
{
	// `SELECT 1 FROM table WHERE bar = $1`
	auto data_exists_result = work.exec_prepared("check_if_data_exists", bar);

	work.commit();

	return data_exists_result.size() == 1;
}

Now, this doesn't work - I get an exception - Started new command 'check_if_data_exists' while command 'check_if_data_exists' was still active, which is kinda disappointing to me, as Im commiting the transaction so it should be ready to start another one right away, but alright.

The question is then, how can I do this to work at all? I'd be glad if stream.complete() itself could return some nice std::expected or something, but sadly the entire API is exception based, Im not sure what to expect.

[tt4g]

The stream_to() is calling the COPY ... FROM STDIN statement, so you cannot call another statement during execution.
Please call data_already_in_database() function with another connection.

• libpqxx/src/stream_to.cxx

  Lines 105 to 114 in 16fe0f0

  	pqxx::stream_to::stream_to(
  	transaction_base &tx, std::string_view path, std::string_view columns) :
  	transaction_focus{tx, s_classname, path},
  	m_finder{pqxx::internal::get_char_finder<
  	'\b', '\f', '\n', '\r', '\t', '\v', '\\'>(
  	pqxx::internal::enc_group(tx.conn().encoding_id()))}
  	{
  	begin_copy(tx, path, columns);
  	register_me();
  	}

• libpqxx/src/stream_to.cxx

  Lines 27 to 35 in 16fe0f0

  	void begin_copy(
  	pqxx::transaction_base &tx, std::string_view table, std::string_view columns)
  	{
  	tx.exec0(
  	std::empty(columns) ?
  	pqxx::internal::concat("COPY "sv, table, " FROM STDIN"sv) :
  	pqxx::internal::concat(
  	"COPY "sv, table, "("sv, columns, ") FROM STDIN"sv));
  	}

[jtv]

Well doing it in another connection may or may not be what you want. But yes, streaming puts the connection in a special state where you can't execute queries. Also, since the stream lives "inside" a transaction, you can't end that transaction before completing the stream.

The only part that's wrong about this is that the error message shouldn't say "while 'check_if_data_exists is still active" — it should refer to the stream as still active instead.

[Dich0tomy]

What would be the best and most idiomatic way to do that, then? Stream in a large amount of data while still checking if any of them is duplicated (and knowing which one is duplicated). Is my only option catching the .complete() call and parsing the postgres error message?

[tt4g]

Can't you use std::set to find duplicate data?

[jtv]

I just opened #880 to fix the error message.

[KayEss]

Your checks are trying to do in data_already_in_database will blow away the performance benefit you may expect to get from the use of stream_to. I think a better way to deal with this is to instead look to use INSERT ... ON CONFLICT and have the database handle the de-duplication for you as you insert the data.

[Dich0tomy]

Yeah but I don't really want to do any action upon conflict. I want to know beforehand if the conflict exists and WHICH field is conflicting, to be able to extract relevant information to present in the diagnostic. And if not beforehand, then simply afterwards, but in a sane way that doesnt introduce partial results on the database and doesnt impose catching and parsing exceptions.

I just now tried to do the same thing with plain exec prepared and Im getting the "started another transaction blah blah" error, which is, well, expected, but Im kinda frustrated that there doesnt seem to be a nice and easy way to do that.

[KayEss]

You can add a tracking field (or more than one) to store the conflicts and then to the INSERT ON CONFLICT. Depending on the fields you get conflicts on you can alter the tracking fields. Once it's all done you can select out any rows with tracking data to make your report, and then UPDATE the tracking fields to reset them back to an empty state.

[Dich0tomy]

That sounds feasible but to not introduce races that would have to be done for each and every table. It also just sounds like a hack. I'm kinda baffled, checking for duplicates/not existing items before doing operations on the database seems like a very natural and ubiquitous operation, I'm really surprised there's no standard and/or nice way of doing that.

[jtv]

In the general case it's possible for someone else to make concurrent changes to be database, invalidating the results of your check. That may explain why it's hard.

One way around it is to work in SERIALIZABLE isolation level. It more or less provides you with an illusion that your transaction is alone on the database. The big downside is that if there's a concurrent change that conflicts with your transaction, the whole thing aborts and you'll have to try again.

Partial results however are typically just not a concern at all, because... transactions.

[Dich0tomy]

That's why I'd hope there's a way to check and write in one transaction/stream, but apparently that's not the case. Thanks for all the help!

[jtv]

By the way @Dich0tomy — more or less the same question also came up in #881.