Rate limit(s) not applied for column hydrate functions

[tho]

Thanks for creating steampipe!

When columns are hydrated the (token bucket) rate limits are not applied.

They were applied when rate limiting support was added in #623 (see 29bb9f0#diff-76a6f7859c8b2e295c2242907a10f928f794bd066f88070e31268b72d68bb11eR101).

However, when support for memoized functions to be assigned as column hydrate functions was removed in #756, the call to h.rateLimit(ctx, d) was lost (see https://github.com/turbot/steampipe-plugin-sdk/pull/756/files#diff-76a6f7859c8b2e295c2242907a10f928f794bd066f88070e31268b72d68bb11eL102).

Therefore, rateLimitDelay in func (h *hydrateCall) start(...) is always 0.

Unless I'm missing something, the fix should be simple:

diff --git a/plugin/hydrate_call.go b/plugin/hydrate_call.go
index 326a90f..37d2df6 100644
--- a/plugin/hydrate_call.go
+++ b/plugin/hydrate_call.go
@@ -93,7 +94,7 @@ func (h *hydrateCall) canStart(rowData *rowData) bool {
 
 // Start starts a hydrate call
 func (h *hydrateCall) start(ctx context.Context, r *rowData, d *QueryData) time.Duration {
-       var rateLimitDelay time.Duration
+       rateLimitDelay := h.rateLimit(ctx, d)
 
        // tell the rowData to wait for this call to complete
        r.wg.Add(1)

I've reproduced the issue with a minimal version of the common List (basic data) / Get (detailed data) pattern described in writing-plugins#about-get-functions with a 1/1 rate limiter:

		RateLimiters: []*rate_limiter.Definition{
			{
				Name:       "test_limiter",
				BucketSize: 1,
				FillRate:   1,
			},
		},

Before patch (unexpected time, 55ms, given value is selected):

$ steampipe query --timing "select id, value from test_table"
+----+-------+
| id | value |
+----+-------+
| 3  | 3     |
| 1  | 1     |
| 2  | 2     |
+----+-------+

Time: 55ms. Rows returned: 0. Rows fetched: 3. Hydrate calls: 3.

After patch (expected time; ~1s per hydrate call => ~3s total):

$ steampipe query --timing "select id, value from test_table"
+----+-------+
| id | value |
+----+-------+
| 1  | 1     |
| 2  | 2     |
| 3  | 3     |
+----+-------+

Time: 3.1s. Rows returned: 0. Rows fetched: 3. Hydrate calls: 3.

plugin.go

func Plugin(context.Context) *plugin.Plugin {
	p := &plugin.Plugin{
		Name:             "steampipe-plugin-test",
		DefaultTransform: transform.FromValue(),
		RateLimiters: []*rate_limiter.Definition{
			{
				Name:       "test_limiter",
				BucketSize: 1,
				FillRate:   1,
			},
		},
		TableMap: map[string]*plugin.Table{
			"test_table": tableTestTable(),
		},
	}

	return p
}

table_test_table.go

package test

import (
	"context"
	"fmt"
	"strconv"

	"github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
	"github.com/turbot/steampipe-plugin-sdk/v5/plugin"
)

func tableTestTable() *plugin.Table {
	return &plugin.Table{
		Name:        "test_table",
		Description: "Test table",
		List: &plugin.ListConfig{
			Hydrate: tableTestTableList,
		},
		Get: &plugin.GetConfig{
			KeyColumns: plugin.SingleColumn("id"),
			Hydrate:    tableTestTableGet,
		},
		Columns: []*plugin.Column{
			{
				Name: "id",
				Type: proto.ColumnType_INT,
			},
			{
				Name:    "value",
				Type:    proto.ColumnType_STRING,
				Hydrate: tableTestTableGet,
			},
		},
	}
}

func tableTestTableList(ctx context.Context, d *plugin.QueryData, _ *plugin.HydrateData) (any, error) {
	ids := []int64{1, 2, 3}

	for _, id := range ids {
		d.StreamListItem(ctx, id)
		if d.RowsRemaining(ctx) == 0 {
			return nil, nil
		}
	}

	return nil, nil
}

func tableTestTableGet(_ context.Context, d *plugin.QueryData, h *plugin.HydrateData) (any, error) {
	if h.Item == nil {
		return d.EqualsQuals["id"].GetInt64Value(), nil
	}

	id, ok := h.Item.(int64)
	if !ok {
		return nil, fmt.Errorf("item is not an ID: %T", h.Item)
	}

	return strconv.FormatInt(id, 10), nil
}