Fix filtering heap scan keys based on index used

When doing an index scan, we should filter out any columns
already used as index scan keys from key columns since they
are later on used to generate heap scan keys.

tsl/src/compression/compression_dml.c

@@ -157,6 +157,16 @@ decompress_batches_for_insert(const ChunkInsertState *cis, TupleTableSlot *slot)
 	if (index_rel)
 		null_columns = NULL;
 
+	if (ts_guc_debug_compression_path_info)
+	{
+		elog(INFO,
+			 "Using %s scan with scan keys: index %d, heap %d, memory %d. ",
+			 index_rel ? "index" : "table",
+			 num_index_scankeys,
+			 num_heap_scankeys,
+			 num_mem_scankeys);
+	}
+
 	/*
 	 * Using latest snapshot to scan the heap since we are doing this to build
 	 * the index on the uncompressed chunks in order to do speculative insertion
@@ -332,20 +342,12 @@ decompress_batch_beginscan(Relation in_rel, Relation index_rel, Snapshot snapsho
 
 	if (index_rel)
 	{
-		if (ts_guc_debug_compression_path_info)
-		{
-			elog(INFO, "Using index scan for DML decompression");
-		}
 		scan->index_scan = index_beginscan(in_rel, index_rel, snapshot, num_scankeys, 0);
 		index_rescan(scan->index_scan, scankeys, num_scankeys, NULL, 0);
 		scan->scan = NULL;
 	}
 	else
 	{
-		if (ts_guc_debug_compression_path_info)
-		{
-			elog(INFO, "Using table scan for DML decompression");
-		}
 		scan->scan = table_beginscan(in_rel, snapshot, num_scankeys, scankeys);
 		scan->index_scan = NULL;
 	}

tsl/src/compression/compression_scankey.c

@@ -322,11 +322,11 @@ build_index_scankeys_using_slot(Oid hypertable_relid, Relation in_rel, Relation
 			AttrNumber idx_attnum = AttrOffsetGetAttrNumber(i);
 			AttrNumber in_attnum = index_rel->rd_index->indkey.values[i];
 			const NameData *attname = attnumAttName(in_rel, in_attnum);
+			AttrNumber column_attno =
+				get_attnum(out_rel->rd_id, NameStr(*attname)) - FirstLowInvalidHeapAttributeNumber;
 
 			/* Make sure we find columns in key columns in order to select the right index */
-			if (!bms_is_member((get_attnum(out_rel->rd_id, NameStr(*attname)) -
-								FirstLowInvalidHeapAttributeNumber),
-							   key_columns))
+			if (!bms_is_member(column_attno, key_columns))
 			{
 				break;
 			}
@@ -337,6 +337,7 @@ build_index_scankeys_using_slot(Oid hypertable_relid, Relation in_rel, Relation
 
 			if (isnull)
 			{
+				*index_columns = bms_add_member(*index_columns, column_attno);
 				ScanKeyEntryInitialize(&scankeys[(*num_scan_keys)++],
 									   SK_ISNULL | SK_SEARCHNULL,
 									   idx_attnum,
@@ -377,6 +378,7 @@ build_index_scankeys_using_slot(Oid hypertable_relid, Relation in_rel, Relation
 			Ensure(OidIsValid(opcode),
 				   "no opcode found for column operator of a hypertable column");
 
+			*index_columns = bms_add_member(*index_columns, column_attno);
 			ScanKeyEntryInitialize(&scankeys[(*num_scan_keys)++],
 								   0, /* flags */
 								   idx_attnum,