Want to do an aggregation using Lua UDF

,
1

I am trying to calculate the number of records where TTL is set to never expire(-1) and the age of the record is more than 45 days. And the count of records where age is less than 45 days and TTL is set to never expire (-1)

I have tried different methods like foreach, stream, aggregate, map but for all cases I am getting this error from Lua UDF. attempt to call a nil value (method 'foreach')"

local function init_counts()
    info("Initializing counts map")
    local m = map()
    m["gt_45_days"] = 0
    m["lt_45_days"] = 0
    return m
end

function count_by_ttl(stream)
    info("***************************Starting count_by_ttl***************************")
    
    -- Create initial counts map
    local counts = init_counts()
    
    local function update_count(rec)
        info("***************************Processing record***************************")
        local currTTL = record.ttl(rec)
        local lut = record.last_update_time(rec)
        local current_time = os.time()
        
        info("***************************Current TTL: %s, LUT: %s***************************", tostring(currTTL), tostring(lut))
        
        if currTTL == 0 then
            local lut_seconds = math.floor(lut / 1000000000)
            local age_in_seconds = current_time - lut_seconds
            local age_in_days = age_in_seconds / (60 * 60 * 24)
            
            info("***************************Record age in days: %s***************************", tostring(age_in_days))
            
            if age_in_days > 45 then
                counts["gt_45_days"] = counts["gt_45_days"] + 1
                info("***************************Record > 45 days old. GT count now: %d***************************", counts["gt_45_days"])
            else
                counts["lt_45_days"] = counts["lt_45_days"] + 1
                info("***************************Record <= 45 days old. LT count now: %d***************************", counts["lt_45_days"])
            end
        else
            info("***************************Skipping record with TTL %s***************************", tostring(currTTL))
        end
    end

    -- Apply the update_count function to each record in the stream
    stream : foreach(update_count)
    
    info("***************************Finished counting. Final counts - GT: %s, LT: %s***************************", 
         tostring(counts["gt_45_days"]), tostring(counts["lt_45_days"]))
    return counts
end

Requesting the community to help me understand the issue. Thank you.