mapped lookup

- change how stride works with gen specs to allow better control over
  how mapped tensors overlap (avoid always overlapping via label "0")

- extend eval fixture verify functionality by adding a function that
  takes param specs directly to verify a single specific case.

15 files changed, 435 insertions, 51 deletions

diff --git a/eval/CMakeLists.txt b/eval/CMakeLists.txt
index 844c81acf52..822a48bffc3 100644
--- a/eval/CMakeLists.txt
+++ b/eval/CMakeLists.txt
@@ -72,6 +72,7 @@ vespa_define_module(
     src/tests/instruction/inplace_map_function
     src/tests/instruction/join_with_number
     src/tests/instruction/l2_distance
+    src/tests/instruction/mapped_lookup
     src/tests/instruction/mixed_112_dot_product
     src/tests/instruction/mixed_inner_product_function
     src/tests/instruction/mixed_simple_join_function
diff --git a/eval/src/tests/eval/gen_spec/gen_spec_test.cpp b/eval/src/tests/eval/gen_spec/gen_spec_test.cpp
index 855b298f295..a52ece90e6a 100644
--- a/eval/src/tests/eval/gen_spec/gen_spec_test.cpp
+++ b/eval/src/tests/eval/gen_spec/gen_spec_test.cpp
@@ -31,13 +31,13 @@ TEST(DimSpecTest, mapped_dimension) {
 
 TEST(DimSpecTest, simple_dictionary_creation) {
     auto dict = DimSpec::make_dict(5, 1, "");
-    std::vector<vespalib::string> expect = {"0", "1", "2", "3", "4"};
+    std::vector<vespalib::string> expect = {"1", "2", "3", "4", "5"};
     EXPECT_EQ(dict, expect);
 }
 
 TEST(DimSpecTest, advanced_dictionary_creation) {
     auto dict = DimSpec::make_dict(5, 3, "str_");
-    std::vector<vespalib::string> expect = {"str_0", "str_3", "str_6", "str_9", "str_12"};
+    std::vector<vespalib::string> expect = {"str_3", "str_6", "str_9", "str_12", "str_15"};
     EXPECT_EQ(dict, expect);
 }
 
@@ -176,50 +176,50 @@ TEST(GenSpecTest, generating_custom_vector) {
 //-----------------------------------------------------------------------------
 
 TensorSpec basic_map = TensorSpec("tensor(a{})")
-    .add({{"a", "0"}}, 1.0)
-    .add({{"a", "1"}}, 2.0)
-    .add({{"a", "2"}}, 3.0);
+    .add({{"a", "1"}}, 1.0)
+    .add({{"a", "2"}}, 2.0)
+    .add({{"a", "3"}}, 3.0);
 
 TensorSpec custom_map = TensorSpec("tensor(a{})")
-    .add({{"a", "s0"}}, 1.0)
-    .add({{"a", "s5"}}, 2.0)
-    .add({{"a", "s10"}}, 3.0);
+    .add({{"a", "s5"}}, 1.0)
+    .add({{"a", "s10"}}, 2.0)
+    .add({{"a", "s15"}}, 3.0);
 
 TEST(GenSpecTest, generating_basic_map) {
     EXPECT_EQ(GenSpec().map("a", 3).gen(), basic_map);
     EXPECT_EQ(GenSpec().map("a", 3, 1).gen(), basic_map);
     EXPECT_EQ(GenSpec().map("a", 3, 1, "").gen(), basic_map);
-    EXPECT_EQ(GenSpec().map("a", {"0", "1", "2"}).gen(), basic_map);
+    EXPECT_EQ(GenSpec().map("a", {"1", "2", "3"}).gen(), basic_map);
 }
 
 TEST(GenSpecTest, generating_custom_map) {
     EXPECT_EQ(GenSpec().map("a", 3, 5, "s").gen(), custom_map);
-    EXPECT_EQ(GenSpec().map("a", {"s0", "s5", "s10"}).gen(), custom_map);
+    EXPECT_EQ(GenSpec().map("a", {"s5", "s10", "s15"}).gen(), custom_map);
 }
 
 //-----------------------------------------------------------------------------
 
 TensorSpec basic_mixed = TensorSpec("tensor(a{},b[1],c{},d[3])")
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 0}}, 1.0)
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 1}}, 2.0)
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 2}}, 3.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 0}}, 4.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 1}}, 5.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 2}}, 6.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 0}}, 7.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 1}}, 8.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 2}}, 9.0);
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 0}}, 1.0)
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 1}}, 2.0)
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 2}}, 3.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 0}}, 4.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 1}}, 5.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 2}}, 6.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 0}}, 7.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 1}}, 8.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 2}}, 9.0);
 
 TensorSpec inverted_mixed = TensorSpec("tensor(a{},b[1],c{},d[3])")
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 0}}, 1.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 0}}, 2.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 0}}, 3.0)
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 1}}, 4.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 1}}, 5.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 1}}, 6.0)
-    .add({{"a", "0"},{"b", 0},{"c", "0"},{"d", 2}}, 7.0)
-    .add({{"a", "1"},{"b", 0},{"c", "0"},{"d", 2}}, 8.0)
-    .add({{"a", "2"},{"b", 0},{"c", "0"},{"d", 2}}, 9.0);
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 0}}, 1.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 0}}, 2.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 0}}, 3.0)
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 1}}, 4.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 1}}, 5.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 1}}, 6.0)
+    .add({{"a", "1"},{"b", 0},{"c", "1"},{"d", 2}}, 7.0)
+    .add({{"a", "2"},{"b", 0},{"c", "1"},{"d", 2}}, 8.0)
+    .add({{"a", "3"},{"b", 0},{"c", "1"},{"d", 2}}, 9.0);
 
 TEST(GenSpecTest, generating_basic_mixed) {
     EXPECT_EQ(GenSpec().map("a", 3).idx("b", 1).map("c", 1).idx("d", 3).gen(), basic_mixed);
diff --git a/eval/src/tests/eval/tensor_lambda/tensor_lambda_test.cpp b/eval/src/tests/eval/tensor_lambda/tensor_lambda_test.cpp
index 657a6ca8d0e..a455e5522b3 100644
--- a/eval/src/tests/eval/tensor_lambda/tensor_lambda_test.cpp
+++ b/eval/src/tests/eval/tensor_lambda/tensor_lambda_test.cpp
@@ -31,7 +31,7 @@ EvalFixture::ParamRepo make_params() {
         .add("x3_float", GenSpec().idx("x", 3).cells(CellType::FLOAT))
         .add("x3_bfloat16", GenSpec().idx("x", 3).cells(CellType::BFLOAT16))
         .add("x3_int8", GenSpec().idx("x", 3).cells(CellType::INT8))
-        .add("x3m", GenSpec().map("x", 3))
+        .add("x3m", GenSpec().map("x", {"0", "1", "2"}))
         .add("x3y5", GenSpec().idx("x", 3).idx("y", 5))
         .add("x3y5_float", GenSpec().idx("x", 3).idx("y", 5).cells(CellType::FLOAT))
         .add("x3y5_bfloat16", GenSpec().idx("x", 3).idx("y", 5).cells(CellType::BFLOAT16))
diff --git a/eval/src/tests/eval/value_type/value_type_test.cpp b/eval/src/tests/eval/value_type/value_type_test.cpp
index 3d4872c2ab3..245c9c30242 100644
--- a/eval/src/tests/eval/value_type/value_type_test.cpp
+++ b/eval/src/tests/eval/value_type/value_type_test.cpp
@@ -353,33 +353,34 @@ TEST("require that dimension index can be obtained") {
 
 void verify_predicates(const ValueType &type,
                        bool expect_error, bool expect_double, bool expect_tensor,
-                       bool expect_sparse, bool expect_dense)
+                       bool expect_sparse, bool expect_dense, bool expect_mixed)
 {
     EXPECT_EQUAL(type.is_error(), expect_error);
     EXPECT_EQUAL(type.is_double(), expect_double);
     EXPECT_EQUAL(type.has_dimensions(), expect_tensor);
     EXPECT_EQUAL(type.is_sparse(), expect_sparse);
     EXPECT_EQUAL(type.is_dense(), expect_dense);
+    EXPECT_EQUAL(type.is_mixed(), expect_mixed);
 }
 
 TEST("require that type-related predicate functions work as expected") {
-    TEST_DO(verify_predicates(type("error"), true, false, false, false, false));
-    TEST_DO(verify_predicates(type("double"), false, true, false, false, false));
-    TEST_DO(verify_predicates(type("tensor()"), false, true, false, false, false));
-    TEST_DO(verify_predicates(type("tensor(x{})"), false, false, true, true, false));
-    TEST_DO(verify_predicates(type("tensor(x{},y{})"), false, false, true, true, false));
-    TEST_DO(verify_predicates(type("tensor(x[5])"), false, false, true, false, true));
-    TEST_DO(verify_predicates(type("tensor(x[5],y[10])"), false, false, true, false, true));
-    TEST_DO(verify_predicates(type("tensor(x[5],y{})"), false, false, true, false, false));
-    TEST_DO(verify_predicates(type("tensor<float>(x{})"), false, false, true, true, false));
-    TEST_DO(verify_predicates(type("tensor<float>(x[5])"), false, false, true, false, true));
-    TEST_DO(verify_predicates(type("tensor<float>(x[5],y{})"), false, false, true, false, false));
-    TEST_DO(verify_predicates(type("tensor<bfloat16>(x{})"), false, false, true, true, false));
-    TEST_DO(verify_predicates(type("tensor<bfloat16>(x[5])"), false, false, true, false, true));
-    TEST_DO(verify_predicates(type("tensor<bfloat16>(x[5],y{})"), false, false, true, false, false));
-    TEST_DO(verify_predicates(type("tensor<int8>(x{})"), false, false, true, true, false));
-    TEST_DO(verify_predicates(type("tensor<int8>(x[5])"), false, false, true, false, true));
-    TEST_DO(verify_predicates(type("tensor<int8>(x[5],y{})"), false, false, true, false, false));
+    TEST_DO(verify_predicates(type("error"), true, false, false, false, false, false));
+    TEST_DO(verify_predicates(type("double"), false, true, false, false, false, false));
+    TEST_DO(verify_predicates(type("tensor()"), false, true, false, false, false, false));
+    TEST_DO(verify_predicates(type("tensor(x{})"), false, false, true, true, false, false));
+    TEST_DO(verify_predicates(type("tensor(x{},y{})"), false, false, true, true, false, false));
+    TEST_DO(verify_predicates(type("tensor(x[5])"), false, false, true, false, true, false));
+    TEST_DO(verify_predicates(type("tensor(x[5],y[10])"), false, false, true, false, true, false));
+    TEST_DO(verify_predicates(type("tensor(x[5],y{})"), false, false, true, false, false, true));
+    TEST_DO(verify_predicates(type("tensor<float>(x{})"), false, false, true, true, false, false));
+    TEST_DO(verify_predicates(type("tensor<float>(x[5])"), false, false, true, false, true, false));
+    TEST_DO(verify_predicates(type("tensor<float>(x[5],y{})"), false, false, true, false, false, true));
+    TEST_DO(verify_predicates(type("tensor<bfloat16>(x{})"), false, false, true, true, false, false));
+    TEST_DO(verify_predicates(type("tensor<bfloat16>(x[5])"), false, false, true, false, true, false));
+    TEST_DO(verify_predicates(type("tensor<bfloat16>(x[5],y{})"), false, false, true, false, false, true));
+    TEST_DO(verify_predicates(type("tensor<int8>(x{})"), false, false, true, true, false, false));
+    TEST_DO(verify_predicates(type("tensor<int8>(x[5])"), false, false, true, false, true, false));
+    TEST_DO(verify_predicates(type("tensor<int8>(x[5],y{})"), false, false, true, false, false, true));
 }
 
 TEST("require that mapped and indexed dimensions can be counted") {
diff --git a/eval/src/tests/instruction/mapped_lookup/CMakeLists.txt b/eval/src/tests/instruction/mapped_lookup/CMakeLists.txt
new file mode 100644
index 00000000000..6bc598235ea
--- /dev/null
+++ b/eval/src/tests/instruction/mapped_lookup/CMakeLists.txt
@@ -0,0 +1,9 @@
+# Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+vespa_add_executable(eval_mapped_lookup_test_app TEST
+    SOURCES
+    mapped_lookup_test.cpp
+    DEPENDS
+    vespaeval
+    GTest::GTest
+)
+vespa_add_test(NAME eval_mapped_lookup_test_app COMMAND eval_mapped_lookup_test_app)
diff --git a/eval/src/tests/instruction/mapped_lookup/mapped_lookup_test.cpp b/eval/src/tests/instruction/mapped_lookup/mapped_lookup_test.cpp
new file mode 100644
index 00000000000..1ffc6b04f4b
--- /dev/null
+++ b/eval/src/tests/instruction/mapped_lookup/mapped_lookup_test.cpp
@@ -0,0 +1,126 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include <vespa/eval/eval/fast_value.h>
+#include <vespa/eval/eval/simple_value.h>
+#include <vespa/eval/instruction/mapped_lookup.h>
+#include <vespa/eval/eval/test/eval_fixture.h>
+#include <vespa/eval/eval/test/gen_spec.h>
+#include <vespa/vespalib/gtest/gtest.h>
+
+using namespace vespalib::eval;
+using namespace vespalib::eval::test;
+
+//-----------------------------------------------------------------------------
+
+struct FunInfo {
+    using LookFor = MappedLookup;
+    bool expect_mutable;
+    FunInfo(bool expect_mutable_in)
+      : expect_mutable(expect_mutable_in) {}
+    void verify(const LookFor &fun) const {
+        EXPECT_EQ(fun.result_is_mutable(), expect_mutable);
+    }
+};
+
+void verify_optimized_cell_types(const vespalib::string &expr) {
+    auto same_stable_types = CellTypeSpace(CellTypeUtils::list_stable_types(), 2).same();
+    auto same_unstable_types = CellTypeSpace(CellTypeUtils::list_unstable_types(), 2).same();
+    auto different_types = CellTypeSpace(CellTypeUtils::list_types(), 2).different();
+    EvalFixture::verify<FunInfo>(expr, {FunInfo(false)}, same_stable_types);
+    EvalFixture::verify<FunInfo>(expr, {}, same_unstable_types);
+    EvalFixture::verify<FunInfo>(expr, {}, different_types);
+}
+
+void verify_optimized(const vespalib::string &expr, bool expect_mutable = false) {
+    CellTypeSpace just_float({CellType::FLOAT}, 2);
+    EvalFixture::verify<FunInfo>(expr, {FunInfo(expect_mutable)}, just_float);
+}
+
+void verify_not_optimized(const vespalib::string &expr) {
+    CellTypeSpace just_float({CellType::FLOAT}, 2);
+    EvalFixture::verify<FunInfo>(expr, {}, just_float);
+}
+
+//-----------------------------------------------------------------------------
+
+TEST(MappedLookup, expression_can_be_optimized) {
+    verify_optimized_cell_types("reduce(x1_1*x5_1y5,sum,x)");
+}
+
+TEST(MappedLookup, key_and_map_can_be_swapped) {
+    verify_optimized("reduce(x5_1y5*x1_1,sum,x)");
+}
+
+TEST(MappedLookup, trivial_indexed_dimensions_are_ignored) {
+    verify_optimized("reduce(c1d1x1_1*a1b1x5_1y5,sum,x,c,d,a,b)");
+    verify_optimized("reduce(c1d1x1_1*a1b1x5_1y5,sum,x,c,a)");
+    verify_optimized("reduce(c1d1x1_1*a1b1x5_1y5,sum,x)");
+}
+
+TEST(MappedLookup, mutable_map_gives_mutable_result) {
+    verify_optimized("reduce(@x1_1*x5_1y5,sum,x)", false);
+    verify_optimized("reduce(x1_1*@x5_1y5,sum,x)", true);
+    verify_optimized("reduce(@x5_1y5*x1_1,sum,x)", true);
+    verify_optimized("reduce(x5_1y5*@x1_1,sum,x)", false);
+    verify_optimized("reduce(@x5_1y5*@x1_1,sum,x)", true);
+}
+
+TEST(MappedLookup, similar_expressions_are_not_optimized) {
+    verify_not_optimized("reduce(x1_1*x5_1,sum,x)");
+    verify_not_optimized("reduce(x1_1*x5_1y5,sum,y)");
+    verify_not_optimized("reduce(x1_1*x5_1y5,sum)");
+    verify_not_optimized("reduce(x1_1*x5_1y5z8,sum,x,y)");
+    verify_not_optimized("reduce(x1_1*x5_1y5,prod,x)");
+    verify_not_optimized("reduce(x1_1y3_3*x5_1y3_2z5,sum,x)");
+    verify_not_optimized("reduce(x1_1y3_3*x5_1y3_2z5,sum,x,y)");
+    verify_not_optimized("reduce(x1_1y5*x5_1z5,sum,x)");
+}
+
+enum class KeyType { EMPTY, UNIT, SCALING, MULTI };
+GenSpec make_key(KeyType type) {
+    switch (type) {
+    case KeyType::EMPTY:   return GenSpec().cells_float().map("x", {});
+    case KeyType::UNIT:    return GenSpec().cells_float().map("x", {"1"}).seq({1.0});
+    case KeyType::SCALING: return GenSpec().cells_float().map("x", {"1"}).seq({5.0});
+    case KeyType::MULTI:   return GenSpec().cells_float().map("x", {"1", "2", "3"}).seq({1.0});
+    }
+    abort();
+}
+
+enum class MapType { EMPTY, SMALL, MEDIUM, LARGE1, LARGE2, LARGE3 };
+GenSpec make_map(MapType type) {
+    switch (type) {
+    case MapType::EMPTY:  return GenSpec().cells_float().idx("y", 5).map("x", {});
+    case MapType::SMALL:  return GenSpec().cells_float().idx("y", 5).map("x", {"1"}).seq(N(10));
+    case MapType::MEDIUM: return GenSpec().cells_float().idx("y", 5).map("x", {"1", "2"}).seq(N(10));
+    case MapType::LARGE1:  return GenSpec().cells_float().idx("y", 5).map("x", 5, 100).seq(N(10));
+    case MapType::LARGE2:  return GenSpec().cells_float().idx("y", 5).map("x", 5, 2).seq(N(10));
+    case MapType::LARGE3:  return GenSpec().cells_float().idx("y", 5).map("x", 5, 1).seq(N(10));
+    }
+    abort();
+}
+
+std::vector<MapType> map_types_for(KeyType key_type) {
+    if (key_type == KeyType::MULTI) {
+        return {MapType::EMPTY, MapType::SMALL, MapType::MEDIUM, MapType::LARGE1, MapType::LARGE2, MapType::LARGE3};
+    } else {
+        return {MapType::EMPTY, MapType::SMALL, MapType::MEDIUM};
+    }
+}
+
+TEST(MappedLookup, test_case_interactions) {
+    for (bool mutable_map: {false, true}) {
+        vespalib::string expr = mutable_map ? "reduce(a*@b,sum,x)" : "reduce(a*b,sum,x)";
+        for (KeyType key_type: {KeyType::EMPTY, KeyType::UNIT, KeyType::SCALING, KeyType::MULTI}) {
+            auto key = make_key(key_type);
+            for (MapType map_type: map_types_for(key_type)) {
+                auto map = make_map(map_type);
+                EvalFixture::verify<FunInfo>(expr, {FunInfo(mutable_map)}, {key,map});
+            }
+        }
+    }
+}
+
+//-----------------------------------------------------------------------------
+
+GTEST_MAIN_RUN_ALL_TESTS()
diff --git a/eval/src/vespa/eval/eval/optimize_tensor_function.cpp b/eval/src/vespa/eval/eval/optimize_tensor_function.cpp
index 63f0f00a13c..41fa550ce07 100644
--- a/eval/src/vespa/eval/eval/optimize_tensor_function.cpp
+++ b/eval/src/vespa/eval/eval/optimize_tensor_function.cpp
@@ -35,7 +35,7 @@
 #include <vespa/eval/instruction/dense_hamming_distance.h>
 #include <vespa/eval/instruction/l2_distance.h>
 #include <vespa/eval/instruction/simple_join_count.h>
-
+#include <vespa/eval/instruction/mapped_lookup.h>
 
 #include <vespa/log/log.h>
 LOG_SETUP(".eval.eval.optimize_tensor_function");
@@ -85,6 +85,7 @@ const TensorFunction &optimize_for_factory(const ValueBuilderFactory &, const Te
                           child.set(MixedInnerProductFunction::optimize(child.get(), stash));
                           child.set(DenseHammingDistance::optimize(child.get(), stash));
                           child.set(SimpleJoinCount::optimize(child.get(), stash));
+                          child.set(MappedLookup::optimize(child.get(), stash));
                       });
     run_optimize_pass(root, [&stash](const Child &child)
                       {
diff --git a/eval/src/vespa/eval/eval/test/eval_fixture.h b/eval/src/vespa/eval/eval/test/eval_fixture.h
index d2244cd4f5f..6cbbfca999b 100644
--- a/eval/src/vespa/eval/eval/test/eval_fixture.h
+++ b/eval/src/vespa/eval/eval/test/eval_fixture.h
@@ -131,6 +131,37 @@ public:
     static const ValueBuilderFactory &test_factory() { return SimpleValueBuilderFactory::get(); }
 
     // Verify the evaluation result and specific tensor function
+    // details for the given expression with the given parameters. A
+    // parameter can be tagged as mutable by giving it a name starting
+    // with '@'. Parameters must be given in automatic discovery order.
+
+    template <typename FunInfo>
+    static void verify(const vespalib::string &expr, const std::vector<FunInfo> &fun_info, std::vector<GenSpec> param_specs) {
+        UNWIND_MSG("in verify(%s) with %zu FunInfo", expr.c_str(), fun_info.size());
+        auto fun = Function::parse(expr);
+        REQUIRE_EQ(fun->num_params(), param_specs.size());
+        EvalFixture::ParamRepo param_repo;
+        for (size_t i = 0; i < fun->num_params(); ++i) {
+            if (fun->param_name(i)[0] == '@') {
+                param_repo.add_mutable(fun->param_name(i), param_specs[i]);
+            } else {
+                param_repo.add(fun->param_name(i), param_specs[i]);
+            }
+        }
+        EvalFixture fixture(prod_factory(), expr, param_repo, true, true);
+        EvalFixture slow_fixture(prod_factory(), expr, param_repo, false, false);
+        EvalFixture test_fixture(test_factory(), expr, param_repo, true, true);
+        REQUIRE_EQ(fixture.result(), test_fixture.result());
+        REQUIRE_EQ(fixture.result(), slow_fixture.result());
+        REQUIRE_EQ(fixture.result(), EvalFixture::ref(expr, param_repo));
+        auto info = fixture.find_all<typename FunInfo::LookFor>();
+        REQUIRE_EQ(info.size(), fun_info.size());
+        for (size_t i = 0; i < fun_info.size(); ++i) {
+            fixture.verify_callback<FunInfo>(fun_info[i], *info[i]);
+        }
+    }
+
+    // Verify the evaluation result and specific tensor function
     // details for the given expression with different combinations of
     // cell types. Parameter names must be valid GenSpec descriptions
     // ('a5b8'), with an optional mutable prefix ('@a5b8') to denote
@@ -138,10 +169,9 @@ public:
     // trailer starting with '$' ('a5b3$2') to allow multiple
     // parameters with the same description as well as scalars
     // ('$this_is_a_scalar').
-                         
+
     template <typename FunInfo>
     static void verify(const vespalib::string &expr, const std::vector<FunInfo> &fun_info, CellTypeSpace cell_type_space) {
-
         UNWIND_MSG("in verify(%s) with %zu FunInfo", expr.c_str(), fun_info.size());
         auto fun = Function::parse(expr);
         REQUIRE_EQ(fun->num_params(), cell_type_space.n());
diff --git a/eval/src/vespa/eval/eval/test/gen_spec.cpp b/eval/src/vespa/eval/eval/test/gen_spec.cpp
index a175111a429..988ff035bbb 100644
--- a/eval/src/vespa/eval/eval/test/gen_spec.cpp
+++ b/eval/src/vespa/eval/eval/test/gen_spec.cpp
@@ -54,7 +54,7 @@ DimSpec::make_dict(size_t size, size_t stride, const vespalib::string &prefix)
 {
     std::vector<vespalib::string> dict;
     for (size_t i = 0; i < size; ++i) {
-        dict.push_back(fmt("%s%zu", prefix.c_str(), i * stride));
+        dict.push_back(fmt("%s%zu", prefix.c_str(), (i + 1) * stride));
     }
     return dict;
 }
diff --git a/eval/src/vespa/eval/eval/test/gen_spec.h b/eval/src/vespa/eval/eval/test/gen_spec.h
index eab2924ac37..5f507eeb272 100644
--- a/eval/src/vespa/eval/eval/test/gen_spec.h
+++ b/eval/src/vespa/eval/eval/test/gen_spec.h
@@ -150,6 +150,7 @@ public:
         _seq = seq_in;
         return *this;
     }
+    GenSpec &seq(const std::vector<double> &numbers) { return seq(Seq({numbers})); }
     bool bad_scalar() const;
     ValueType type() const;
     TensorSpec gen() const;
diff --git a/eval/src/vespa/eval/eval/value_type.cpp b/eval/src/vespa/eval/eval/value_type.cpp
index 80e62b5f512..dc5ce645a8c 100644
--- a/eval/src/vespa/eval/eval/value_type.cpp
+++ b/eval/src/vespa/eval/eval/value_type.cpp
@@ -191,6 +191,18 @@ ValueType::is_dense() const
     return true;
 }
 
+bool
+ValueType::is_mixed() const
+{
+    bool seen_mapped = false;
+    bool seen_indexed = false;
+    for (const auto &dim : dimensions()) {
+        seen_mapped |= dim.is_mapped();
+        seen_indexed |= dim.is_indexed();
+    }
+    return (seen_mapped && seen_indexed);
+}
+
 size_t
 ValueType::count_indexed_dimensions() const
 {
diff --git a/eval/src/vespa/eval/eval/value_type.h b/eval/src/vespa/eval/eval/value_type.h
index 0822fa1e4b0..b7a7c92e137 100644
--- a/eval/src/vespa/eval/eval/value_type.h
+++ b/eval/src/vespa/eval/eval/value_type.h
@@ -60,6 +60,7 @@ public:
     bool has_dimensions() const { return !_dimensions.empty(); }
     bool is_sparse() const;
     bool is_dense() const;
+    bool is_mixed() const;
     size_t count_indexed_dimensions() const;
     size_t count_mapped_dimensions() const;
     size_t dense_subspace_size() const;
diff --git a/eval/src/vespa/eval/instruction/CMakeLists.txt b/eval/src/vespa/eval/instruction/CMakeLists.txt
index 2146e3ee8ab..7df3f745e79 100644
--- a/eval/src/vespa/eval/instruction/CMakeLists.txt
+++ b/eval/src/vespa/eval/instruction/CMakeLists.txt
@@ -31,6 +31,7 @@ vespa_add_library(eval_instruction OBJECT
     inplace_map_function.cpp
     join_with_number_function.cpp
     l2_distance.cpp
+    mapped_lookup.cpp
     mixed_112_dot_product.cpp
     mixed_inner_product_function.cpp
     mixed_simple_join_function.cpp
diff --git a/eval/src/vespa/eval/instruction/mapped_lookup.cpp b/eval/src/vespa/eval/instruction/mapped_lookup.cpp
new file mode 100644
index 00000000000..31d3c9766bf
--- /dev/null
+++ b/eval/src/vespa/eval/instruction/mapped_lookup.cpp
@@ -0,0 +1,152 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "mapped_lookup.h"
+#include <vespa/eval/eval/wrap_param.h>
+#include <vespa/eval/eval/fast_value.hpp>
+#include <vespa/vespalib/util/require.h>
+
+namespace vespalib::eval {
+
+using namespace tensor_function;
+using namespace operation;
+using namespace instruction;
+
+namespace {
+
+template <typename CT>
+ConstArrayRef<CT> my_mapped_lookup_fallback(const Value::Index &key_idx, const Value::Index &map_idx,
+                                            const CT *key_cells, const CT *map_cells, size_t res_size, Stash &stash) __attribute__((noinline));
+template <typename CT>
+ConstArrayRef<CT> my_mapped_lookup_fallback(const Value::Index &key_idx, const Value::Index &map_idx,
+                                            const CT *key_cells, const CT * map_cells, size_t res_size, Stash &stash)
+{
+    SparseJoinPlan plan(1);
+    auto result = stash.create_array<CT>(res_size);
+    SparseJoinState sparse(plan, key_idx, map_idx);
+    auto outer = sparse.first_index.create_view({});
+    auto inner = sparse.second_index.create_view(sparse.second_view_dims);
+    outer->lookup({});
+    while (outer->next_result(sparse.first_address, sparse.first_subspace)) {
+        inner->lookup(sparse.address_overlap);
+        if (inner->next_result(sparse.second_only_address, sparse.second_subspace)) {
+            auto factor = key_cells[sparse.lhs_subspace];
+            const CT *match = map_cells + (res_size * sparse.rhs_subspace);
+            for (size_t i = 0; i < result.size(); ++i) {
+                result[i] += factor * match[i];
+            }
+        }
+    }
+    return result;
+}
+
+template <typename CT>
+struct MappedLookupResult {
+    ArrayRef<CT> value;
+    MappedLookupResult(size_t res_size, Stash &stash)
+      : value(stash.create_array<CT>(res_size)) {}
+    void process_match(CT factor, const CT *match) {
+        for (size_t i = 0; i < value.size(); ++i) {
+            value[i] += factor * match[i];
+        }
+    }
+};
+
+template <typename CT>
+ConstArrayRef<CT> my_fast_mapped_lookup(const FastAddrMap &key_map, const FastAddrMap &map_map,
+                                        const CT *key_cells, const CT *map_cells, size_t res_size, Stash &stash)
+{
+    if ((key_map.size() == 1) && (key_cells[0] == 1.0)) {
+        auto subspace = map_map.lookup_singledim(key_map.labels()[0]);
+        if (subspace != FastAddrMap::npos()) {
+            return {map_cells + (res_size * subspace), res_size};
+        } else {
+            return stash.create_array<CT>(res_size);
+        }
+    }
+    MappedLookupResult<CT> result(res_size, stash);
+    if (key_map.size() <= map_map.size()) {
+        const auto &labels = key_map.labels();
+        for (size_t i = 0; i < labels.size(); ++i) {
+            auto subspace = map_map.lookup_singledim(labels[i]);
+            if (subspace != FastAddrMap::npos()) {
+                result.process_match(key_cells[i], map_cells + (res_size * subspace));
+            }
+        }
+    } else {
+        const auto &labels = map_map.labels();
+        for (size_t i = 0; i < labels.size(); ++i) {
+            auto subspace = key_map.lookup_singledim(labels[i]);
+            if (subspace != FastAddrMap::npos()) {
+                result.process_match(key_cells[subspace], map_cells + (res_size * i));
+            }
+        }
+    }
+    return result.value;
+}
+
+template <typename CT>
+void my_mapped_lookup_op(InterpretedFunction::State &state, uint64_t param) {
+    const auto &res_type = unwrap_param<ValueType>(param);
+    const auto &key_idx = state.peek(1).index();
+    const auto &map_idx = state.peek(0).index();
+    const CT *key_cells = state.peek(1).cells().typify<CT>().cbegin();
+    const CT *map_cells = state.peek(0).cells().typify<CT>().cbegin();
+    auto result = __builtin_expect(are_fast(key_idx, map_idx), true)
+        ? my_fast_mapped_lookup<CT>(as_fast(key_idx).map, as_fast(map_idx).map, key_cells, map_cells, res_type.dense_subspace_size(), state.stash)
+        : my_mapped_lookup_fallback<CT>(key_idx, map_idx, key_cells, map_cells, res_type.dense_subspace_size(), state.stash);
+    state.pop_pop_push(state.stash.create<DenseValueView>(res_type, TypedCells(result)));
+}
+
+bool check_types(const ValueType &res, const ValueType &key, const ValueType &map) {
+    return ((res.is_dense()) && (key.dense_subspace_size() == 1) && (map.is_mixed()) &&
+            (res.cell_type() == key.cell_type()) &&
+            (res.cell_type() == map.cell_type()) &&
+            ((res.cell_type() == CellType::FLOAT) || (res.cell_type() == CellType::DOUBLE)) &&
+            (key.mapped_dimensions().size() == 1) &&
+            (key.mapped_dimensions() == map.mapped_dimensions()) &&
+            (map.nontrivial_indexed_dimensions() == res.nontrivial_indexed_dimensions()));
+}
+
+} // namespace <unnamed>
+
+MappedLookup::MappedLookup(const ValueType &res_type,
+                           const TensorFunction &key_in,
+                           const TensorFunction &map_in)
+  : tensor_function::Op2(res_type, key_in, map_in)
+{
+}
+
+InterpretedFunction::Instruction
+MappedLookup::compile_self(const ValueBuilderFactory &, Stash &) const
+{
+    uint64_t param = wrap_param<ValueType>(result_type());
+    if (result_type().cell_type() == CellType::FLOAT) {
+        return {my_mapped_lookup_op<float>, param};
+    }
+    if (result_type().cell_type() == CellType::DOUBLE) {
+        return {my_mapped_lookup_op<double>, param};
+    }
+    REQUIRE_FAILED("cell types must be float or double");
+}
+
+const TensorFunction &
+MappedLookup::optimize(const TensorFunction &expr, Stash &stash)
+{
+    auto reduce = as<Reduce>(expr);
+    if (reduce && (reduce->aggr() == Aggr::SUM)) {
+        auto join = as<Join>(reduce->child());
+        if (join && (join->function() == Mul::f)) {
+            const TensorFunction &lhs = join->lhs();
+            const TensorFunction &rhs = join->rhs();
+            if (check_types(expr.result_type(), lhs.result_type(), rhs.result_type())) {
+                return stash.create<MappedLookup>(expr.result_type(), lhs, rhs);
+            }
+            if (check_types(expr.result_type(), rhs.result_type(), lhs.result_type())) {
+                return stash.create<MappedLookup>(expr.result_type(), rhs, lhs);
+            }
+        }
+    }
+    return expr;
+}
+
+} // namespace
diff --git a/eval/src/vespa/eval/instruction/mapped_lookup.h b/eval/src/vespa/eval/instruction/mapped_lookup.h
new file mode 100644
index 00000000000..2b66c6cf4aa
--- /dev/null
+++ b/eval/src/vespa/eval/instruction/mapped_lookup.h
@@ -0,0 +1,49 @@
+// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#pragma once
+
+#include <vespa/eval/eval/tensor_function.h>
+
+namespace vespalib::eval {
+
+/**
+ * Tensor function implementing generalized lookup of 'key' in 'map'
+ * with some type restrictions.
+ *
+ * 'key' may only contain the lookup dimension (called 'x' here)
+ * 'map' must have full mapped overlap with 'key'
+ *
+ * Both tensors must have the same cell type, which can be either
+ * float or double.
+ * 
+ * The optimized expression looks like this: reduce(key*map,sum,x)
+ *
+ * If 'map' is also sparse, the lookup operation is a sparse dot
+ * product and will be optimized using SparseDotProductFunction
+ * instead.
+ *
+ * The best performance (simple hash lookup with a result referencing
+ * existing cells without having to copy them) is achieved when a
+ * single dense subspace in 'map' matches a cell with value 1.0 from
+ * 'key'. This fast-path can be ensured if this optimization is
+ * combined with the simple_join_count optimization:
+ *
+ * key = tensor(x{}):{my_key:1}
+ * map = tensor(x{},y[128])
+ * fallback = tensor(y[128])
+ *
+ * simple lookup with fallback:
+ * if(reduce(key*map,count)==128,reduce(key*map,sum,x),fallback)
+ **/
+class MappedLookup : public tensor_function::Op2
+{
+public:
+    MappedLookup(const ValueType &res_type, const TensorFunction &key_in, const TensorFunction &map_in);
+    const TensorFunction &key() const { return lhs(); }
+    const TensorFunction &map() const { return rhs(); }
+    InterpretedFunction::Instruction compile_self(const ValueBuilderFactory &factory, Stash &stash) const override;
+    bool result_is_mutable() const override { return map().result_is_mutable(); }
+    static const TensorFunction &optimize(const TensorFunction &expr, Stash &stash);
+};
+
+} // namespace