Fix crashes with malformed compressed data

The Gorilla and Dictionary compressors crashed on malformed data
with the reverse decompression iterator and the bulk decompressor.

This issue was reported by Mehmet Ince @mdisec https://mehmetince.net/,
and the test case binaries are his contribution that I integrated into
the fuzzer test suite.

This change adds validation for these cases and adds reverse iterator
coverage to the fuzzer tests, so the randomised test will catch future
issues. It already caught issues in other compressors like deltadelta,
which I also fixed in this review.

(cherry picked from commit 517c13e)

Author: dbeck

.unreleased/pr_10360

@@ -0,0 +1,2 @@
+Fixes: #10360 Decompressor crashes with malformed compressed data
+Thanks: @mdisec for reporting the issues with the validation of the compressed data during decompression

src/adts/bit_array.h

@@ -37,6 +37,7 @@ pg_attribute_always_inline static uint64 bit_array_iter_next(BitArrayIterator *i
 static void bit_array_iterator_init_rev(BitArrayIterator *iter, const BitArray *array);
 /* return last num_bits in forward-order (not reverse-order); must have been written as num_bits */
 static uint64 bit_array_iter_next_rev(BitArrayIterator *iter, uint8 num_bits);
+static inline uint64 bit_array_iter_position(const BitArrayIterator *iter);
 
 /* I/O */
 static inline void bit_array_send(StringInfo buffer, const BitArray *data);

src/adts/bit_array_impl.h

@@ -85,8 +85,10 @@ bit_array_num_buckets(const BitArray *array)
 static inline uint64
 bit_array_num_bits(const BitArray *array)
 {
-	return (BITS_PER_BUCKET * (array->buckets.num_elements - UINT64CONST(1))) +
-		   array->bits_used_in_last_bucket;
+	return array->buckets.num_elements == 0 ?
+			   0 :
+			   (BITS_PER_BUCKET * (array->buckets.num_elements - UINT64CONST(1))) +
+				   array->bits_used_in_last_bucket;
 }
 
 static inline uint64 *
@@ -250,6 +252,15 @@ bit_array_iterator_init(BitArrayIterator *iter, const BitArray *array)
 	};
 }
 
+static inline uint64
+bit_array_iter_position(const BitArrayIterator *iter)
+{
+	Assert(iter->current_bucket >= 0);
+	return iter->current_bucket < 0 ?
+			   0 :
+			   (BITS_PER_BUCKET * iter->current_bucket) + iter->bits_used_in_current_bucket;
+}
+
 pg_attribute_always_inline static uint64
 bit_array_iter_next(BitArrayIterator *iter, uint8 num_bits)
 {
@@ -302,7 +313,7 @@ bit_array_iterator_init_rev(BitArrayIterator *iter, const BitArray *array)
 {
 	*iter = (BitArrayIterator){
 		.array = array,
-		.current_bucket = array->buckets.num_elements - 1,
+		.current_bucket = array->buckets.num_elements == 0 ? 0 : array->buckets.num_elements - 1,
 		.bits_used_in_current_bucket = array->bits_used_in_last_bucket,
 	};
 }

tsl/src/compression/algorithms/deltadelta.c

@@ -60,6 +60,11 @@ typedef struct DeltaDeltaDecompressionIterator
 	Simple8bRleDecompressionIterator delta_deltas;
 	Simple8bRleDecompressionIterator nulls;
 	bool has_nulls;
+
+	/* for validating the last_value set in the header
+	 * and cross check with the iterator */
+	uint64 last_value_to_return;
+	uint64 last_returned_value;
 } DeltaDeltaDecompressionIterator;
 
 typedef struct DeltaDeltaCompressor
@@ -481,6 +486,44 @@ delta_delta_compressor_append_value(DeltaDeltaCompressor *compressor, int64 next
 /**********************************************************************************/
 /**********************************************************************************/
 
+static inline uint32
+decompression_iterator_items_seen(const DeltaDeltaDecompressionIterator *iter)
+{
+	if (iter->has_nulls)
+	{
+		return iter->nulls.num_elements_returned;
+	}
+	else
+	{
+		return iter->delta_deltas.num_elements_returned;
+	}
+}
+
+static inline uint32
+decompression_iterator_item_count(const DeltaDeltaDecompressionIterator *iter)
+{
+	if (iter->has_nulls)
+	{
+		return iter->nulls.num_elements;
+	}
+	else
+	{
+		return iter->delta_deltas.num_elements;
+	}
+}
+
+static inline uint32
+decompression_iterator_values_seen(const DeltaDeltaDecompressionIterator *iter)
+{
+	return iter->delta_deltas.num_elements_returned;
+}
+
+static inline uint32
+decompression_iterator_value_count(const DeltaDeltaDecompressionIterator *iter)
+{
+	return iter->delta_deltas.num_elements;
+}
+
 static void
 int64_decompression_iterator_init_forward(DeltaDeltaDecompressionIterator *iter, void *compressed,
 										  Oid element_type)
@@ -507,6 +550,8 @@ int64_decompression_iterator_init_forward(DeltaDeltaDecompressionIterator *iter,
 		.prev_val = 0,
 		.prev_delta = 0,
 		.has_nulls = has_nulls,
+		.last_returned_value = 0,
+		.last_value_to_return = header->last_value,
 	};
 
 	simple8brle_decompression_iterator_init_forward(&iter->delta_deltas, deltas);
@@ -515,9 +560,13 @@ int64_decompression_iterator_init_forward(DeltaDeltaDecompressionIterator *iter,
 	{
 		Simple8bRleSerialized *nulls = bytes_deserialize_simple8b_and_advance(&si);
 		simple8brle_decompression_iterator_init_forward(&iter->nulls, nulls);
+		CheckCompressedData(deltas->num_elements <= nulls->num_elements);
 	}
 }
 
+static DecompressResultInternal
+delta_delta_decompression_iterator_try_next_forward_internal(DeltaDeltaDecompressionIterator *iter);
+
 static void
 int64_decompression_iterator_init_reverse(DeltaDeltaDecompressionIterator *iter, void *compressed,
 										  Oid element_type)
@@ -529,7 +578,7 @@ int64_decompression_iterator_init_reverse(DeltaDeltaDecompressionIterator *iter,
 	DeltaDeltaCompressed *header = consumeCompressedData(&si, sizeof(DeltaDeltaCompressed));
 	Simple8bRleSerialized *deltas = bytes_deserialize_simple8b_and_advance(&si);
 
-	Assert(header->has_nulls == 0 || header->has_nulls == 1);
+	CheckCompressedData(header->has_nulls == 0 || header->has_nulls == 1);
 
 	*iter = (DeltaDeltaDecompressionIterator){
 		.base = {
@@ -541,6 +590,8 @@ int64_decompression_iterator_init_reverse(DeltaDeltaDecompressionIterator *iter,
 		.prev_val = header->last_value,
 		.prev_delta = header->last_delta,
 		.has_nulls = header->has_nulls,
+		.last_returned_value = 0,
+		.last_value_to_return = 0,
 	};
 
 	simple8brle_decompression_iterator_init_reverse(&iter->delta_deltas, deltas);
@@ -549,6 +600,31 @@ int64_decompression_iterator_init_reverse(DeltaDeltaDecompressionIterator *iter,
 	{
 		Simple8bRleSerialized *nulls = bytes_deserialize_simple8b_and_advance(&si);
 		simple8brle_decompression_iterator_init_reverse(&iter->nulls, nulls);
+		CheckCompressedData(deltas->num_elements <= nulls->num_elements);
+	}
+
+	/* on reverse iteration the `last_value` in the header is critical for
+	 * the rest of the decoding. when we receive malformed data we can only
+	 * cross reference the result with the first value
+	 */
+	{
+		DeltaDeltaDecompressionIterator forward_iter;
+		forward_iter.base.compression_algorithm = COMPRESSION_ALGORITHM_DELTADELTA;
+		forward_iter.base.forward = true;
+		forward_iter.base.element_type = element_type;
+		forward_iter.base.try_next = delta_delta_decompression_iterator_try_next_forward;
+		int64_decompression_iterator_init_forward(&forward_iter, compressed, element_type);
+		/* find the first non-null value */
+		DecompressResultInternal first_val =
+			delta_delta_decompression_iterator_try_next_forward_internal(&forward_iter);
+		int32 n = 0;
+		while (first_val.is_done == false && first_val.is_null)
+		{
+			CheckCompressedData(n++ < GLOBAL_MAX_ROWS_PER_COMPRESSION);
+			first_val = delta_delta_decompression_iterator_try_next_forward_internal(&forward_iter);
+		}
+		CheckCompressedData(!first_val.is_done);
+		iter->last_value_to_return = first_val.val;
 	}
 }
 
@@ -615,6 +691,14 @@ delta_delta_decompression_iterator_try_next_forward_internal(DeltaDeltaDecompres
 			simple8brle_decompression_iterator_try_next_forward(&iter->nulls);
 		if (result.is_done)
 		{
+			/* make sure we exhausted all items before  */
+			CheckCompressedData(decompression_iterator_items_seen(iter) ==
+								decompression_iterator_item_count(iter));
+			/* and also that we returned all values */
+			CheckCompressedData(decompression_iterator_values_seen(iter) ==
+								decompression_iterator_value_count(iter));
+			/* the last element must match the expected value */
+			CheckCompressedData(iter->last_returned_value == iter->last_value_to_return);
 			return (DecompressResultInternal){
 				.is_done = true,
 			};
@@ -633,6 +717,14 @@ delta_delta_decompression_iterator_try_next_forward_internal(DeltaDeltaDecompres
 
 	if (result.is_done)
 	{
+		/* make sure we exhausted all items before  */
+		CheckCompressedData(decompression_iterator_items_seen(iter) ==
+							decompression_iterator_item_count(iter));
+		/* and also that we returned all values */
+		CheckCompressedData(decompression_iterator_values_seen(iter) ==
+							decompression_iterator_value_count(iter));
+		/* the last element must match the expected value */
+		CheckCompressedData(iter->last_returned_value == iter->last_value_to_return);
 		return (DecompressResultInternal){
 			.is_done = true,
 		};
@@ -642,6 +734,7 @@ delta_delta_decompression_iterator_try_next_forward_internal(DeltaDeltaDecompres
 
 	iter->prev_delta += delta_delta;
 	iter->prev_val += iter->prev_delta;
+	iter->last_returned_value = iter->prev_val;
 
 	return (DecompressResultInternal){
 		.val = iter->prev_val,
@@ -712,14 +805,22 @@ delta_delta_decompression_iterator_try_next_reverse_internal(DeltaDeltaDecompres
 			simple8brle_decompression_iterator_try_next_reverse(&iter->nulls);
 		if (result.is_done)
 		{
+			/* make sure we exhausted all items before  */
+			CheckCompressedData(decompression_iterator_items_seen(iter) ==
+								decompression_iterator_item_count(iter));
+			/* and also that we returned all values */
+			CheckCompressedData(decompression_iterator_values_seen(iter) ==
+								decompression_iterator_value_count(iter));
+			/* the last element must match the expected value */
+			CheckCompressedData(iter->last_returned_value == iter->last_value_to_return);
 			return (DecompressResultInternal){
 				.is_done = true,
 			};
 		}
 
 		if (result.val != 0)
 		{
-			Assert(result.val == 1);
+			CheckCompressedData(result.val == 1);
 			return (DecompressResultInternal){
 				.is_null = true,
 			};
@@ -730,6 +831,14 @@ delta_delta_decompression_iterator_try_next_reverse_internal(DeltaDeltaDecompres
 
 	if (result.is_done)
 	{
+		/* make sure we exhausted all items before  */
+		CheckCompressedData(decompression_iterator_items_seen(iter) ==
+							decompression_iterator_item_count(iter));
+		/* and also that we returned all values */
+		CheckCompressedData(decompression_iterator_values_seen(iter) ==
+							decompression_iterator_value_count(iter));
+		/* the last element must match the expected value */
+		CheckCompressedData(iter->last_returned_value == iter->last_value_to_return);
 		return (DecompressResultInternal){
 			.is_done = true,
 		};
@@ -740,6 +849,7 @@ delta_delta_decompression_iterator_try_next_reverse_internal(DeltaDeltaDecompres
 	delta_delta = zig_zag_decode(result.val);
 	iter->prev_val -= iter->prev_delta;
 	iter->prev_delta -= delta_delta;
+	iter->last_returned_value = val;
 
 	return (DecompressResultInternal){
 		.val = val,

tsl/src/compression/algorithms/deltadelta_impl.c

@@ -90,6 +90,7 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 #undef INNER_LOOP_SIZE
 
 	uint64 *restrict validity_bitmap = NULL;
+	ELEMENT_TYPE last_value = decompressed_values[n_total - 1];
 	if (has_nulls)
 	{
 		/* Now move the data to account for nulls, and fill the validity bitmap. */
@@ -117,6 +118,7 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 		CheckCompressedData(n_notnull + simple8brle_bitmap_num_ones(&nulls) == n_total);
 
 		int current_notnull_element = n_notnull - 1;
+		last_value = decompressed_values[current_notnull_element];
 		for (int i = n_total - 1; i >= 0; i--)
 		{
 			Assert(i >= current_notnull_element);
@@ -136,6 +138,9 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 		Assert(current_notnull_element == -1);
 	}
 
+	/* the last value in the header must match with the last that we returned */
+	CheckCompressedData(last_value == (ELEMENT_TYPE) header->last_value);
+
 	/* Return the result. */
 	ArrowArray *result = MemoryContextAllocZero(dest_mctx, sizeof(ArrowArray) + sizeof(void *) * 2);
 	const void **buffers = (const void **) &result[1];