Record manipulation with more than one filter - lua

lua
udf
java

#1

Hi,

I have a range query with three restrictions. i wrote a StreamUDF with one filter function for every condition and it didn´t work.

The query:

SELECT id1, id2, link_type, visibility, data, time, version FROM linktable
WHERE id1 = <id1> AND
  link_type = <link_type> AND
  time >= <minTime> AND
  time <= <maxTimestamp> AND
  visibility = VISIBILITY_DEFAULT
ORDER BY time DESC LIMIT <offset>, <limit>;

the java code to invoke this function:

stmt = new Statement();
stmt.setNamespace(dbid);
stmt.setSetName("links");
stmt.setIndexName("time");
stmt.setFilters(Filter.range("time", minTimestamp, maxTimestamp));
stmt.setAggregateFunction("linkbench", "check_id1", Value.get(id1));
stmt.setAggregateFunction("linkbench", "check_linktype", Value.get(link_type));
resultSet = client.queryAggregate(null, stmt, "linkbench", "check_visibility", Value.get(VISIBILITY_DEFAULT));

Lua script:

local function map_links(record)
    -- Add user and password to returned map.
    -- Could add other record bins here as well.
    return record.id2
end

function check_id1(stream,id1)
    local function filter_id1(record)
        return record.id1 == id1
    end
    return stream : filter(filter_id1) : map(map_links)
end

function check_linktype(stream,link_type)
    local function filter_linktype(record)
        return record.link_type == link_type
    end
    return stream : filter(filter_linktype) : map(map_links)
end

function check_visibility(stream,visibility)
    local function filter_visibility(record)
        return record.visibility == visibility
    end
    return stream : filter(filter_visibility) : map(map_links)
end

is there any solution to filter the stream record in one lua function with 3 filters ? this will be than called with three parameters in java (Value.get()). Is there any tutorial ?


Client queries using multiple filters
#2

You have to have all your filter functions chained together. Here is a sample data set I created for your problem to illustrate the solution, followed by the UDF and the results. I tested via AQL.

pgupta@ubuntu:~/testProjects/discuss3637$ aql -f "./runscript" RUN './runscript’ remove module "filterRecords.lua" OK, 1 module removed.

register module './filterRecords.lua’ OK, 1 module added.

select * from test.links ±—±-----±----------±-----------±---------+ | id | time | link_type | visibility | data | ±—±-----±----------±-----------±---------+ | 1 | 1590 | 0 | 1 | “abc81” | | 3 | 1570 | 1 | 1 | “abc41” | | 2 | 1540 | 0 | 0 | “abc12” | | 1 | 1550 | 1 | 0 | “abc221” | | 3 | 1520 | 0 | 1 | “abc11” | | 2 | 1560 | 0 | 0 | “abc32” | | 2 | 1530 | 1 | 0 | “abc92” | | 1 | 1580 | 1 | 0 | “abc52” | | 3 | 1510 | 0 | 1 | “abc2” | | 1 | 1500 | 1 | 1 | “abc1” | ±—±-----±----------±-----------±---------+ 10 rows in set (0.047 secs)

AGGREGATE filterRecords.check_visibility(1,0,3,1515,1545) ON test.links ±-----------------------------------------------------------------------------+ | check_visib… | ±-----------------------------------------------------------------------------+ | MAP(’{“data”:“abc11”, “time”:1520, “visibility”:1, “id”:3, “link_type”:0}’) | ±-----------------------------------------------------------------------------+ 1 row in set (0.064 secs)

Here is the UDF used:

pgupta@ubuntu:~/testProjects/discuss3637$ more filterRecords.lua
local function rec_to_map(rec)
   local xrec = map()
   for i, bin_name in ipairs(record.bin_names(rec)) do
       xrec[bin_name] = rec[bin_name]
   end
   return xrec
end

function check_visibility(stream, vis, linktype, id, mintime, maxtime)
   local function range_filter(rec)
      local val = rec['time']
      if val >= mintime and val <= maxtime then
         return true
      else
         return false
      end
   end
   local function idcheck(rec)
      local val = rec['id']
      if val == id then
         return true
      else
         return false
      end
   end
   local function linktypecheck(rec)
      local val = rec['link_type']
      if val == linktype then
         return true
      else
         return false
      end
   end
   local function vischeck(rec)
      local val = rec['visibility']
      if val == vis then
         return true
      else
         return false
      end
   end

   return stream:filter(range_filter):filter(idcheck):filter(linktypecheck):filter(vischeck):map(rec_to_map)
end

Note: You could have used a secondary index on timestamp and filtered the initial stream in your query Statement being fed to the stream udf. Your java implementation should either specify one stream udf in the setAggregationFuntion() or call one stream UDF in queryAggregate() - use the correct overloaded function.

Hope this helps!