@Override
    public DataBag exec(Tuple input) throws IOException {
        if ( input == null || input.size() == 0 || input.get(0) == null ) {
            return null;
        }

        DataBag bag = mBagFactory.newDefaultBag();
        try {
            String haystack = (String) input.get(0);
            Iterator<SearchResult<String>> results = searcher.search(haystack.getBytes(UTF8));
            while ( results.hasNext() ) {
                Set<String> outputs = results.next().getOutputs();
                for ( String output : outputs ) {
                    bag.add(mTupleFactory.newTuple(output));
                }
            }
        } catch (Throwable t ) {
            t.printStackTrace();
            throw WrappedIOException.wrap("Could not process tuple: " + input, t);
        }
        return bag;
    }