Use arrow cache only for index scans

Since compressed segments are accessed sequentially during
sequence/columnar scans, it is not necessary to cache the
decompressed arrow arrays.

Instead, use the arrow cache only if an index scan is detected. Also
make it possible to manually disable the cache by setting the GUC
hypercore_arrow_cache_max_entries to 0.

Author: erimatnor

src/guc.c

@@ -1120,12 +1120,13 @@ _guc_init(void)
 							/* short_desc= */ "max number of entries in arrow data cache",
 							/* long_desc= */
 							"The max number of decompressed arrow segments that can be "
-							"cached before entries are evicted. This mainly affects the "
-							"performance of index scans on the Hypercore TAM "
-							"when segments are accessed in non-sequential order.",
+							"cached before entries are evicted. This only affects the "
+							"performance of index scans when using Hypercore TAM "
+							"and segments are accessed in non-sequential order. "
+							"Set to 0 to disable the use of the cache.",
 							/* valueAddr= */ &ts_guc_hypercore_arrow_cache_max_entries,
 							/* bootValue= */ 25000,
-							/* minValue= */ 1,
+							/* minValue= */ 0,
 							/* maxValue= */ INT_MAX,
 							/* context= */ PGC_USERSET,
 							/* flags= */ 0,

tsl/src/hypercore/arrow_array.c

@@ -141,6 +141,15 @@ arrow_release_buffers(ArrowArray *array)
 		arrow_private_release(array);
 }
 
+void
+arrow_release(ArrowArray *array)
+{
+	if (array->release != NULL)
+		array->release(array);
+
+	pfree(array);
+}
+
 /*
  * Variable-size primitive layout ArrowArray from decompression iterator.
  */

tsl/src/hypercore/arrow_array.h

@@ -15,3 +15,4 @@ extern NullableDatum arrow_get_datum(const ArrowArray *array, Oid typid, int16 t
 									 uint16 index);
 extern ArrowArray *arrow_from_compressed(Datum compressed, Oid typid, MemoryContext dest_mcxt,
 										 MemoryContext tmp_mcxt);
+extern void arrow_release(ArrowArray *array);

tsl/src/hypercore/arrow_cache.c

@@ -174,13 +174,7 @@ arrow_cache_clear_entry(ArrowColumnCacheEntry *restrict entry)
 		if (entry->arrow_arrays[i])
 		{
 			ArrowArray *array = entry->arrow_arrays[i];
-
-			if (array->release)
-			{
-				array->release(array);
-				array->release = NULL;
-			}
-			pfree(array);
+			arrow_release(array);
 			entry->arrow_arrays[i] = NULL;
 		}
 	}

tsl/src/hypercore/arrow_tts.c

@@ -4,6 +4,7 @@
  * LICENSE-TIMESCALE for a copy of the license.
  */
 #include <postgres.h>
+#include "guc.h"
 #include <access/attnum.h>
 #include <access/htup_details.h>
 #include <access/tupdesc.h>
@@ -94,6 +95,7 @@ tts_arrow_init(TupleTableSlot *slot)
 	aslot->total_row_count = 0;
 	aslot->referenced_attrs = NULL;
 	aslot->arrow_qual_result = NULL;
+	aslot->arrow_arrays = NULL;
 
 	/*
 	 * Set up child slots, one for the non-compressed relation and one for the
@@ -112,6 +114,8 @@ tts_arrow_init(TupleTableSlot *slot)
 	aslot->child_slot = aslot->noncompressed_slot;
 	aslot->valid_attrs = palloc0(sizeof(bool) * slot->tts_tupleDescriptor->natts);
 	aslot->segmentby_attrs = palloc0(sizeof(bool) * slot->tts_tupleDescriptor->natts);
+	aslot->arrow_arrays = palloc0(sizeof(ArrowArray *) * slot->tts_tupleDescriptor->natts);
+
 	/* Note that aslot->referenced_attrs is initialized on demand, and not
 	 * here, because NULL is a valid state for referenced_attrs. */
 	MemoryContextSwitchTo(oldmcxt);
@@ -128,6 +132,24 @@ tts_arrow_init(TupleTableSlot *slot)
 		GenerationContextCreateCompat(slot->tts_mcxt, "Per-segment memory context", 64 * 1024);
 }
 
+static void
+clear_arrow_arrays(TupleTableSlot *slot)
+{
+	ArrowTupleTableSlot *aslot = (ArrowTupleTableSlot *) slot;
+
+	if (aslot->arrow_arrays)
+	{
+		for (int i = 0; i < slot->tts_tupleDescriptor->natts; i++)
+		{
+			if (aslot->arrow_arrays[i] != NULL)
+			{
+				arrow_release(aslot->arrow_arrays[i]);
+				aslot->arrow_arrays[i] = NULL;
+			}
+		}
+	}
+}
+
 /*
  * The release function is called by:
  *
@@ -153,6 +175,7 @@ tts_arrow_release(TupleTableSlot *slot)
 	aslot->compressed_slot = NULL;
 	aslot->noncompressed_slot = NULL;
 	aslot->arrow_cache_entry = NULL;
+	aslot->arrow_arrays = NULL;
 }
 
 static void
@@ -269,13 +292,15 @@ tts_arrow_clear(TupleTableSlot *slot)
 	memset(aslot->valid_attrs, 0, sizeof(bool) * slot->tts_tupleDescriptor->natts);
 	aslot->arrow_cache_entry = NULL;
 	aslot->arrow_qual_result = NULL;
+	clear_arrow_arrays(slot);
 	MemoryContextReset(aslot->per_segment_mcxt);
 }
 
 static inline void
 tts_arrow_store_tuple(TupleTableSlot *slot, TupleTableSlot *child_slot, uint16 tuple_index)
 {
 	ArrowTupleTableSlot *aslot = (ArrowTupleTableSlot *) slot;
+	bool clear_arrow_data = true;
 
 	Assert(!TTS_EMPTY(child_slot));
 	Assert(OidIsValid(slot->tts_tableOid));
@@ -315,6 +340,8 @@ tts_arrow_store_tuple(TupleTableSlot *slot, TupleTableSlot *child_slot, uint16 t
 
 				if (!ItemPointerEquals(&decoded_tid, &child_slot->tts_tid))
 					clear_arrow_parent(slot);
+				else
+					clear_arrow_data = false;
 			}
 		}
 
@@ -339,6 +366,10 @@ tts_arrow_store_tuple(TupleTableSlot *slot, TupleTableSlot *child_slot, uint16 t
 	aslot->child_slot = child_slot;
 	aslot->tuple_index = tuple_index;
 	aslot->arrow_cache_entry = NULL;
+
+	if (clear_arrow_data)
+		clear_arrow_arrays(slot);
+
 	/* Clear valid attributes */
 	memset(aslot->valid_attrs, 0, sizeof(bool) * slot->tts_tupleDescriptor->natts);
 	MemoryContextReset(aslot->per_segment_mcxt);
@@ -462,6 +493,52 @@ is_compressed_col(const TupleDesc tupdesc, AttrNumber attno)
 	return coltypid == typinfo->type_oid;
 }
 
+static inline ArrowArray *
+get_arrow_array(ArrowTupleTableSlot *aslot, const int16 attoff)
+{
+	const AttrNumber attnum = AttrOffsetGetAttrNumber(attoff);
+	TupleTableSlot *slot = &aslot->base.base;
+
+	/*
+	 * Only use the arrow array cache if the slot is used in an index scan and
+	 * the cache hasn't been disabled by configuration.
+	 */
+	if (aslot->index_attrs != NULL && ts_guc_hypercore_arrow_cache_max_entries > 0)
+	{
+		ArrowArray **arrow_arrays = arrow_column_cache_read_one(aslot, attnum);
+		return arrow_arrays[attoff];
+	}
+
+	Assert(aslot->arrow_arrays);
+
+	if (NULL == aslot->arrow_arrays[attoff])
+	{
+		const int16 *attrs_offset_map = arrow_slot_get_attribute_offset_map(&aslot->base.base);
+		const AttrNumber cattno = AttrOffsetGetAttrNumber(attrs_offset_map[attoff]);
+		const TupleDesc compressed_tupdesc = aslot->compressed_slot->tts_tupleDescriptor;
+
+		if (is_compressed_col(compressed_tupdesc, cattno))
+		{
+			bool isnull;
+			Datum value = slot_getattr(aslot->child_slot, cattno, &isnull);
+
+			/* Can this ever be NULL? */
+			if (!isnull)
+			{
+				const ArrowColumnCache *acache = &aslot->arrow_cache;
+				const TupleDesc tupdesc = slot->tts_tupleDescriptor;
+				const Form_pg_attribute attr = TupleDescAttr(tupdesc, attoff);
+				aslot->arrow_arrays[attoff] = arrow_from_compressed(value,
+																	attr->atttypid,
+																	slot->tts_mcxt,
+																	acache->decompression_mcxt);
+			}
+		}
+	}
+
+	return aslot->arrow_arrays[attoff];
+}
+
 static pg_attribute_always_inline ArrowArray *
 set_attr_value(TupleTableSlot *slot, const int16 attoff)
 {
@@ -486,13 +563,11 @@ set_attr_value(TupleTableSlot *slot, const int16 attoff)
 	}
 	else
 	{
-		const AttrNumber attnum = AttrOffsetGetAttrNumber(attoff);
-		ArrowArray **arrow_arrays = arrow_column_cache_read_one(aslot, attnum);
-
-		arrow_array = arrow_arrays[attoff];
+		arrow_array = get_arrow_array(aslot, attoff);
 
 		if (arrow_array == NULL)
 		{
+			const AttrNumber attnum = AttrOffsetGetAttrNumber(attoff);
 			/* Since the column is not the segment-by column, and there is no
 			 * decompressed data, the column must be NULL. Use the default
 			 * value. */
@@ -506,7 +581,7 @@ set_attr_value(TupleTableSlot *slot, const int16 attoff)
 			const Oid typid = attr->atttypid;
 			const int16 typlen = attr->attlen;
 			const NullableDatum datum =
-				arrow_get_datum(arrow_arrays[attoff], typid, typlen, aslot->tuple_index - 1);
+				arrow_get_datum(arrow_array, typid, typlen, aslot->tuple_index - 1);
 			slot->tts_values[attoff] = datum.value;
 			slot->tts_isnull[attoff] = datum.isnull;
 		}
@@ -777,7 +852,6 @@ arrow_slot_get_array(TupleTableSlot *slot, AttrNumber attno)
 {
 	ArrowTupleTableSlot *aslot = (ArrowTupleTableSlot *) slot;
 	const int attoff = AttrNumberGetAttrOffset(attno);
-	ArrowArray **arrow_arrays;
 
 	TS_DEBUG_LOG("attno: %d, tuple_index: %d", attno, aslot->tuple_index);
 
@@ -800,8 +874,7 @@ arrow_slot_get_array(TupleTableSlot *slot, AttrNumber attno)
 	if (!aslot->valid_attrs[attoff])
 		return set_attr_value(slot, attoff);
 
-	arrow_arrays = arrow_column_cache_read_one(aslot, attno);
-	return arrow_arrays[attoff];
+	return get_arrow_array(aslot, attoff);
 }
 
 /*

tsl/src/hypercore/arrow_tts.h

@@ -73,6 +73,7 @@ typedef struct ArrowTupleTableSlot
 	uint16 total_row_count;
 	ArrowColumnCache arrow_cache;
 	ArrowColumnCacheEntry *arrow_cache_entry;
+	ArrowArray **arrow_arrays;
 	bool *referenced_attrs;
 	bool *segmentby_attrs;
 	bool *valid_attrs;		 /* Per-column validity up to "tts_nvalid" */

tsl/test/expected/hypercore_columnar.out

@@ -107,8 +107,7 @@ $$, :'chunk'));
          Scankey: (device < 4)
          Vectorized Filter: (location = 2)
          Rows Removed by Filter: 16
- Array: cache misses=N, decompress count=N calls=N
-(6 rows)
+(5 rows)
 
 -- Save away all data from the chunk so that we can compare.
 create table saved as select * from :chunk;
@@ -139,8 +138,7 @@ $$, :'chunk'));
    ->  Custom Scan (ColumnarScan) on _hyper_I_N_chunk (actual rows=N loops=N)
          Vectorized Filter: (humidity > '110'::double precision)
          Rows Removed by Filter: 204
- Array: cache misses=N, decompress count=N calls=N
-(5 rows)
+(4 rows)
 
 select count(*) from :chunk where humidity > 110;
  count 
@@ -159,8 +157,7 @@ $$, :'chunk'));
    ->  Custom Scan (ColumnarScan) on _hyper_I_N_chunk (actual rows=N loops=N)
          Vectorized Filter: (humidity > '50'::double precision)
          Rows Removed by Filter: 87
- Array: cache misses=N, decompress count=N calls=N
-(5 rows)
+(4 rows)
 
 select lhs.count, rhs.count
 from (select count(*) from :chunk where humidity > 50) lhs,
@@ -191,8 +188,7 @@ $$, :'chunk'));
    ->  Custom Scan (ColumnarScan) on _hyper_I_N_chunk (actual rows=N loops=N)
          Filter: (temp > '50'::numeric)
          Rows Removed by Filter: 204
- Array: cache misses=N, decompress count=N calls=N
-(5 rows)
+(4 rows)
 
 select count(*) from :chunk where temp > 50;
  count 
@@ -210,8 +206,7 @@ $$, :'chunk'));
    ->  Custom Scan (ColumnarScan) on _hyper_I_N_chunk (actual rows=N loops=N)
          Filter: (temp > '20'::numeric)
          Rows Removed by Filter: 98
- Array: cache misses=N, decompress count=N calls=N
-(5 rows)
+(4 rows)
 
 select lhs.count, rhs.count
 from (select count(*) from :chunk where temp > 20) lhs,
@@ -242,8 +237,7 @@ select count(*) from :chunk where humidity > 40 and temp > 20;
          Filter: (temp > '20'::numeric)
          Rows Removed by Filter: 132
          Vectorized Filter: (humidity > '40'::double precision)
- Array: cache misses=30, decompress count=60 calls=165
-(6 rows)
+(5 rows)
 
 select count(*) from :chunk where humidity > 40 and temp > 20;
  count 
@@ -272,8 +266,7 @@ $$, :'chunk'));
          Rows Removed by Filter: 3
          Scankey: (device = 3)
          Vectorized Filter: (humidity > '40'::double precision)
- Array: cache misses=N, decompress count=N calls=N
-(7 rows)
+(6 rows)
 
 select count(*) from :chunk where humidity > 40 and temp > 20 and device = 3;
  count 
@@ -303,8 +296,7 @@ $$, :'chunk'));
          ->  Seq Scan on _hyper_I_N_chunk (actual rows=N loops=N)
                Filter: (device < 4)
                Rows Removed by Filter: 184
- Array: cache misses=N, decompress count=N calls=N
-(8 rows)
+(7 rows)
 
 drop table readings;
 drop table saved;

tsl/test/expected/hypercore_index_btree.out

@@ -435,8 +435,7 @@ $$, :'chunk1'));
 ------------------------------------------------------------------------
  Custom Scan (ColumnarScan) on _hyper_I_N_chunk (actual rows=N loops=N)
    Scankey: ((location_id >= 5) AND (location_id <= 10))
- Array: cache misses=N, decompress count=N calls=N
-(3 rows)
+(2 rows)
 
 -- These should generate decompressions as above, but for all columns.
 select explain_analyze_anonymize(format($$

tsl/test/expected/hypercore_scans.out

@@ -377,8 +377,7 @@ on conflict (location, device, time) do nothing;
          ->  Custom Scan (ChunkDispatch) (actual rows=8641 loops=1)
                ->  Subquery Scan on "*SELECT*" (actual rows=8641 loops=1)
                      ->  Function Scan on generate_series t (actual rows=8641 loops=1)
- Array: cache misses=2, decompress count=4 calls=4
-(10 rows)
+(9 rows)
 
 -- This should show values for all columns
 explain (analyze, costs off, timing off, summary off, decompress_cache_stats)
@@ -447,8 +446,7 @@ order by time desc;
          Scankey: (location = '1'::text)
          Vectorized Filter: (location = '1'::text)
          Rows Removed by Filter: 113
- Array: cache misses=27, decompress count=81 calls=239
-(8 rows)
+(7 rows)
 
 --  Save the data for comparison with seqscan
 create temp table chunk_saved as
@@ -526,8 +524,7 @@ select count(*) from :chunk where location = 1::text;
          Scankey: (location = '1'::text)
          Vectorized Filter: (location = '1'::text)
          Rows Removed by Filter: 113
- Array: cache misses=27, decompress count=27 calls=27
-(6 rows)
+(5 rows)
 
 -- Testing same thing with SeqScan. It still decompresses in the
 -- count(*) case, although it shouldn't have to. So, probably an
@@ -541,8 +538,7 @@ select count(*) from :chunk where device = 1;
    ->  Seq Scan on _hyper_1_1_chunk (actual rows=17 loops=1)
          Filter: (device = 1)
          Rows Removed by Filter: 392
- Array: cache misses=30, decompress count=62 calls=410
-(5 rows)
+(4 rows)
 
 explain (analyze, costs off, timing off, summary off, decompress_cache_stats)
 select device from :chunk where device = 1;
@@ -561,8 +557,7 @@ select count(*) from :chunk where location = 1::text;
    ->  Seq Scan on _hyper_1_1_chunk (actual rows=89 loops=1)
          Filter: (location = '1'::text)
          Rows Removed by Filter: 320
- Array: cache misses=30, decompress count=62 calls=410
-(5 rows)
+(4 rows)
 
 -- ColumnarScan declares itself as projection capable. This query
 -- would add a Result node on top if ColumnarScan couldn't project.