From c2f040fa0aea8f11f0bef4d3699bb56633f8166d Mon Sep 17 00:00:00 2001
From: Sven Klemm <sven@timescale.com>
Date: Sat, 1 Feb 2025 14:12:45 +0100
Subject: [PATCH] Fix DELETE on compressed chunk with non-btree operators

Fixes #7644
---
 tsl/src/compression/compression_dml.c     | 14 ++++++++--
 tsl/src/compression/compression_dml.h     |  2 +-
 tsl/src/compression/compression_scankey.c | 32 +++++++++++++++--------
 3 files changed, 34 insertions(+), 14 deletions(-)

diff --git a/tsl/src/compression/compression_dml.c b/tsl/src/compression/compression_dml.c
index 573039b3ca5..ba160e7b1c7 100644
--- a/tsl/src/compression/compression_dml.c
+++ b/tsl/src/compression/compression_dml.c
@@ -268,7 +268,8 @@ decompress_batches_for_update_delete(HypertableModifyState *ht_state, Chunk *chu
 		scankeys = build_update_delete_scankeys(comp_chunk_rel,
 												heap_filters,
 												&num_scankeys,
-												&null_columns);
+												&null_columns,
+												&delete_only);
 	}
 
 	if (matching_index_rel)
@@ -957,7 +958,16 @@ process_predicates(Chunk *ch, CompressionSettings *settings, List *predicates,
 							break;
 						}
 						default:
-							/* Do nothing for unknown operator strategies. */
+							*heap_filters = lappend(*heap_filters,
+													make_batchfilter(column_name,
+																	 op_strategy,
+																	 collation,
+																	 opcode,
+																	 arg_value,
+																	 false, /* is_null_check */
+																	 false, /* is_null */
+																	 false	/* is_array_op */
+																	 ));
 							break;
 					}
 					continue;
diff --git a/tsl/src/compression/compression_dml.h b/tsl/src/compression/compression_dml.h
index 90b77ce73f0..d30ed3f6610 100644
--- a/tsl/src/compression/compression_dml.h
+++ b/tsl/src/compression/compression_dml.h
@@ -45,4 +45,4 @@ ScanKeyData *build_heap_scankeys(Oid hypertable_relid, Relation in_rel, Relation
 								 CompressionSettings *settings, Bitmapset *key_columns,
 								 Bitmapset **null_columns, TupleTableSlot *slot, int *num_scankeys);
 ScanKeyData *build_update_delete_scankeys(Relation in_rel, List *heap_filters, int *num_scankeys,
-										  Bitmapset **null_columns);
+										  Bitmapset **null_columns, bool *delete_only);
diff --git a/tsl/src/compression/compression_scankey.c b/tsl/src/compression/compression_scankey.c
index be5e6ef9652..a816a33ffc8 100644
--- a/tsl/src/compression/compression_scankey.c
+++ b/tsl/src/compression/compression_scankey.c
@@ -435,7 +435,7 @@ build_index_scankeys_using_slot(Oid hypertable_relid, Relation in_rel, Relation
  */
 ScanKeyData *
 build_update_delete_scankeys(Relation in_rel, List *heap_filters, int *num_scankeys,
-							 Bitmapset **null_columns)
+							 Bitmapset **null_columns, bool *delete_only)
 {
 	ListCell *lc;
 	BatchFilter *filter;
@@ -455,16 +455,26 @@ build_update_delete_scankeys(Relation in_rel, List *heap_filters, int *num_scank
 							NameStr(filter->column_name),
 							RelationGetRelationName(in_rel))));
 
-		key_index = create_segment_filter_scankey(in_rel,
-												  NameStr(filter->column_name),
-												  filter->strategy,
-												  deduce_filter_subtype(filter, typoid),
-												  scankeys,
-												  key_index,
-												  null_columns,
-												  filter->value ? filter->value->constvalue : 0,
-												  filter->is_null_check,
-												  filter->is_array_op);
+		int new_index = create_segment_filter_scankey(in_rel,
+													  NameStr(filter->column_name),
+													  filter->strategy,
+													  deduce_filter_subtype(filter, typoid),
+													  scankeys,
+													  key_index,
+													  null_columns,
+													  filter->value ? filter->value->constvalue : 0,
+													  filter->is_null_check,
+													  filter->is_array_op);
+		/*
+		 * When we plan to DELETE directly on compressed chunks we
+		 * need to ensure all query constraints could be applied
+		 * to the compressed scan and disable direct DELETE when
+		 * we are skipping filters.
+		 */
+		if (*delete_only && new_index == key_index)
+			*delete_only = false;
+
+		key_index = new_index;
 	}
 	*num_scankeys = key_index;
 	return scankeys;