Fix leak during concurrent UPDATE/DELETE

[sb230132]

When updating and deleting the same tuple while both transactions are running at the same time, we end up with reference leak. This is because one of the query in a transaction fails and we take error path. However we fail to close the table.

This patch fixes the above mentioned problem by closing the required tables.

Fixes #5674

Disable-check: force-changelog-changed

[github-actions]

@lkshminarayanan, @shhnwz: please review this pull request.

Powered by pull-review

[codecov]

Codecov Report

Merging #5675 (fec4840) into main (656daf4) will decrease coverage by 0.03%.
The diff coverage is 50.00%.

@@            Coverage Diff             @@
##             main    #5675      +/-   ##
==========================================
- Coverage   90.99%   90.97%   -0.03%     
==========================================
  Files         230      230              
  Lines       54573    54579       +6     
==========================================
- Hits        49659    49653       -6     
- Misses       4914     4926      +12     

Impacted Files	Coverage Δ
tsl/src/compression/compression.c	95.53% <50.00%> (-0.38%)	⬇️

... and 7 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[kgyrtkirk]

I've already noticed this issue without even a test in this comment

however ; the PR was merged regardless of my comments or followup being created ?!@!

I think we have serious flaws in answering comments/acting on them...

[kgyrtkirk]

Could we have some documentation about the return value contract of this method?
That might be important; as there are callsites relying on it - see my previous comment in the other PR for an example usage

[sb230132]

I have documented it like this:

/*
 * This method will:
 *  1.scan compressed chunk
 *  2.decompress the row
 *  3.delete this row from compressed chunk
 *  4.insert decompressed rows to uncompressed chunk
 *
 * Return value:
 * if all 4 steps defined above pass set chunk_status_changed to true and return true
 * if step 3 fails return false. Step 3 will fail if there are conflicting concurrent operations on same chunk.
 */
static bool
decompress_batches(RowDecompressor *decompressor, ScanKeyData *scankeys, int num_scankeys,
				   Bitmapset *null_columns, List *is_nulls, bool *chunk_status_changed)

[kgyrtkirk]

I think that's just a comment; about the operation and nothing to do with documentation

I don't think that could be accepted as a contract - it just exposes the complexity of the method...(leaving most likely the current implementation the only one which could satisfy that)

I think it should be some clear-as-day statement like:

Decompresses all rows satisfying the provided scankeys

It will return true if it have decompressed matching rows

[kgyrtkirk]

this seems like unreachable code to me - isn't it?

[sb230132]

For Repeatable Read Isolation level this code gets executed.

[antekresic]

Lets put the table_endscan before the error and I suppose we don't need the return statements.

[sb230132]

I've already noticed this issue without even a test in this comment

however ; the PR was merged regardless of my comments or followup being created ?!@!

I think we have serious flaws in answering comments/acting on them...

My bad, will take care of it from now.

[antekresic]

Lets add a bit documentation on the cases we are handling here.

[kgyrtkirk]

that's nonsense....if it can't happen - why handle it at all?

[antekresic]

Suggested change

	case TM_BeingModified:
	/* TM_BeingModified cannot happen since we always wait for locks here
	* TM_SelfModified should never happen
	*/
	case TM_BeingModified:

[kgyrtkirk]

that's nonsense....if it can't happen - why handle it at all?

[kgyrtkirk]

I think the following is true:

• the method will return true iff it processed all candidate rows
• chunk_status_changed will be true iff there are rows decompressed

note:
because it only return true or an error right now - I think chunk_status_changed is not necessary anymore...

[fabriziomello]

LGTM!

It's up to you but I'm in favor of some separation of concerns in the code to make it easier to understand.

My main concern where is that we have a while loop and inside a for with some specific logic to skip tuples and then we have a switch with another specific logic and both logic decides if we continue the while and all this mix of continue and break in different contexts make the code a bit confusing so my suggestion below is to extract both logics into small check functions, something like:

diff --git a/tsl/src/compression/compression.c b/tsl/src/compression/compression.c
index f11359898..512fdde10 100644
--- a/tsl/src/compression/compression.c
+++ b/tsl/src/compression/compression.c
@@ -2261,6 +2261,78 @@ build_update_delete_scankeys(RowDecompressor *decompressor, List *filters, int *
 	return scankeys;
 }
 
+static bool
+decompress_batches_skip_tuple(HeapTuple compressed_tuple, TupleDesc in_desc,
+							  Bitmapset *null_columns, List *is_nulls)
+{
+	bool skip_tuple = false;
+	int attrno = bms_next_member(null_columns, -1);
+	int pos = 0;
+	bool is_null_condition = 0;
+	bool seg_col_is_null = false;
+	for (; attrno >= 0; attrno = bms_next_member(null_columns, attrno))
+	{
+		is_null_condition = list_nth_int(is_nulls, pos);
+		seg_col_is_null = heap_attisnull(compressed_tuple, attrno, in_desc);
+		if ((seg_col_is_null && !is_null_condition) || (!seg_col_is_null && is_null_condition))
+		{
+			/*
+			 * if segment by column in the scanned tuple has non null value
+			 * and IS NULL is specified, OR segment by column has null value
+			 * and IS NOT NULL is specified then skip this tuple
+			 */
+			skip_tuple = true;
+			break;
+		}
+		pos++;
+	}
+
+	return skip_tuple;
+}
+
+static bool
+decompress_batches_tuple_deleted(TM_Result result, TableScanDesc heapScan)
+{
+	bool tuple_deleted = false;
+
+	switch (result)
+	{
+		/* If the tuple has been already deleted, most likely somebody
+		 * decompressed the tuple already */
+		case TM_Deleted:
+			if (IsolationUsesXactSnapshot())
+			{
+				/* For Repeatable Read isolation level report error */
+				table_endscan(heapScan);
+				ereport(ERROR,
+						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+						 errmsg("could not serialize access due to concurrent update")));
+			}
+			break;
+		/*
+		 * If another transaction is updating the compressed data,
+		 * we have to abort the transaction to keep consistency.
+		 */
+		case TM_Updated:
+			table_endscan(heapScan);
+			elog(ERROR, "tuple concurrently updated");
+			break;
+		case TM_Invisible:
+			table_endscan(heapScan);
+			elog(ERROR, "attempted to lock invisible tuple");
+			break;
+		case TM_Ok:
+			tuple_deleted = true;
+			break;
+		default:
+			table_endscan(heapScan);
+			elog(ERROR, "unexpected tuple operation result: %d", result);
+			break;
+	}
+
+	return tuple_deleted;
+}
+
 /*
  * This method will:
  *  1.scan compressed chunk
@@ -2282,38 +2354,23 @@ decompress_batches(RowDecompressor *decompressor, ScanKeyData *scankeys, int num
 
 	TableScanDesc heapScan =
 		table_beginscan(decompressor->in_rel, snapshot, num_scankeys, scankeys);
+
 	while ((compressed_tuple = heap_getnext(heapScan, ForwardScanDirection)) != NULL)
 	{
-		bool skip_tuple = false;
-		int attrno = bms_next_member(null_columns, -1);
-		int pos = 0;
-		bool is_null_condition = 0;
-		bool seg_col_is_null = false;
-		for (; attrno >= 0; attrno = bms_next_member(null_columns, attrno))
-		{
-			is_null_condition = list_nth_int(is_nulls, pos);
-			seg_col_is_null = heap_attisnull(compressed_tuple, attrno, decompressor->in_desc);
-			if ((seg_col_is_null && !is_null_condition) || (!seg_col_is_null && is_null_condition))
-			{
-				/*
-				 * if segment by column in the scanned tuple has non null value
-				 * and IS NULL is specified, OR segment by column has null value
-				 * and IS NOT NULL is specified then skip this tuple
-				 */
-				skip_tuple = true;
-				break;
-			}
-			pos++;
-		}
-		if (skip_tuple)
+		TM_FailureData tmfd;
+		TM_Result result;
+
+		if (decompress_batches_skip_tuple(compressed_tuple,
+										  decompressor->in_desc,
+										  null_columns,
+										  is_nulls))
 			continue;
+
 		heap_deform_tuple(compressed_tuple,
 						  decompressor->in_desc,
 						  decompressor->compressed_datums,
 						  decompressor->compressed_is_nulls);
 
-		TM_FailureData tmfd;
-		TM_Result result;
 		result = table_tuple_delete(decompressor->in_rel,
 									&compressed_tuple->t_self,
 									decompressor->mycid,
@@ -2323,54 +2380,17 @@ decompress_batches(RowDecompressor *decompressor, ScanKeyData *scankeys, int num
 									&tmfd,
 									false);
 
-		switch (result)
-		{
-			/* If the tuple has been already deleted, most likely somebody
-			 * decompressed the tuple already */
-			case TM_Deleted:
-			{
-				if (IsolationUsesXactSnapshot())
-				{
-					/* For Repeatable Read isolation level report error */
-					table_endscan(heapScan);
-					ereport(ERROR,
-							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
-							 errmsg("could not serialize access due to concurrent update")));
-				}
-				continue;
-			}
-			break;
-			/*
-			 * If another transaction is updating the compressed data,
-			 * we have to abort the transaction to keep consistency.
-			 */
-			case TM_Updated:
-			{
-				table_endscan(heapScan);
-				elog(ERROR, "tuple concurrently updated");
-			}
-			break;
-			case TM_Invisible:
-			{
-				table_endscan(heapScan);
-				elog(ERROR, "attempted to lock invisible tuple");
-			}
-			break;
-			case TM_Ok:
-				break;
-			default:
-			{
-				table_endscan(heapScan);
-				elog(ERROR, "unexpected tuple operation result: %d", result);
-			}
-			break;
-		}
+		if (!decompress_batches_tuple_deleted(result, heapScan))
+			continue;
+
 		row_decompressor_decompress_row(decompressor, NULL);
 		*chunk_status_changed = true;
 	}
+
 	if (scankeys)
 		pfree(scankeys);
 	table_endscan(heapScan);
+
 	return true;
 }

[sb230132]

Thanks Fabrizio for this comment. As you mentioned, this function seem more confusing and likely error prone in future. Its the same concern even Zoltan has and we plan to open a new issue to refactor this code and do some cleanup as well as part of the new PR.
So for now i would like to keep it as it is and fix in new PR. Hope you agree.

[timescale-automation]

Automated backport to 2.10.x not done: cherry-pick failed.

Git status

HEAD detached at origin/2.10.x
You are currently cherry-picking commit 2d71a5bc.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	both modified:   tsl/src/compression/compression.c
	deleted by us:   tsl/test/isolation/expected/compression_dml_iso.out
	deleted by us:   tsl/test/isolation/specs/compression_dml_iso.spec

no changes added to commit (use "git add" and/or "git commit -a")

Job log

[antekresic]

@sb230132 can you create the refactor issue and link to this PR for posterity?

Thanks.