diff --git a/quesma/schema/registry.go b/quesma/schema/registry.go
index 4374423be..38f05a6ce 100644
--- a/quesma/schema/registry.go
+++ b/quesma/schema/registry.go
@@ -43,7 +43,7 @@ func (s *schemaRegistry) loadSchemas() (map[TableName]Schema, error) {
 		s.populateSchemaFromStaticConfiguration(indexConfiguration, fields)
 		s.populateSchemaFromTableDefinition(definitions, indexName, fields)
 		s.populateAliases(indexConfiguration, fields, aliases)
-		schemas[TableName(indexName)] = Schema{Fields: fields, Aliases: aliases}
+		schemas[TableName(indexName)] = NewSchemaWithAliases(fields, aliases)
 	}
 
 	return schemas, nil
diff --git a/quesma/schema/registry_test.go b/quesma/schema/registry_test.go
index 9789f74db..0b5536134 100644
--- a/quesma/schema/registry_test.go
+++ b/quesma/schema/registry_test.go
@@ -42,11 +42,10 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchema(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}),
 			exists: true,
 		},
 		{
@@ -64,11 +63,10 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchema(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}),
 			exists: true,
 		},
 		{
@@ -85,11 +83,10 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchema(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}),
 			exists: true,
 		},
 		{
@@ -110,11 +107,10 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchema(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}),
 			exists: true,
 		},
 		{
@@ -130,10 +126,8 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 			},
 			tableDiscovery: fixedTableProvider{tables: map[string]schema.Table{}},
 			tableName:      "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
-				"message": {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
-			exists: true,
+			want:           schema.NewSchema(map[schema.FieldName]schema.Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword}}),
+			exists:         true,
 		},
 		{
 			name: "schema inferred, with mapping overrides",
@@ -154,11 +148,10 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				},
 				}}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchema(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}),
 			exists: true,
 		},
 		{
@@ -181,13 +174,12 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchemaWithAliases(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{
-					"message_alias": "message",
-				}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}, map[schema.FieldName]schema.FieldName{
+				"message_alias": "message",
+			}),
 			exists: true,
 		},
 		{
@@ -208,13 +200,12 @@ func Test_schemaRegistry_FindSchema(t *testing.T) {
 				}},
 			}},
 			tableName: "some_table",
-			want: schema.Schema{Fields: map[schema.FieldName]schema.Field{
+			want: schema.NewSchemaWithAliases(map[schema.FieldName]schema.Field{
 				"message":    {PropertyName: "message", InternalPropertyName: "message", Type: schema.TypeKeyword},
 				"event_date": {PropertyName: "event_date", InternalPropertyName: "event_date", Type: schema.TypeTimestamp},
-				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}},
-				Aliases: map[schema.FieldName]schema.FieldName{
-					"message_alias": "message",
-				}},
+				"count":      {PropertyName: "count", InternalPropertyName: "count", Type: schema.TypeLong}}, map[schema.FieldName]schema.FieldName{
+				"message_alias": "message",
+			}),
 			exists: true,
 		},
 		{
diff --git a/quesma/schema/schema.go b/quesma/schema/schema.go
index 338d2c8f5..d6d1d0422 100644
--- a/quesma/schema/schema.go
+++ b/quesma/schema/schema.go
@@ -4,8 +4,9 @@ package schema
 
 type (
 	Schema struct {
-		Fields  map[FieldName]Field
-		Aliases map[FieldName]FieldName
+		Fields              map[FieldName]Field
+		Aliases             map[FieldName]FieldName
+		internalNameToField map[FieldName]Field
 	}
 	Field struct {
 		// PropertyName is how users refer to the field
@@ -18,6 +19,22 @@ type (
 	FieldName string
 )
 
+func NewSchemaWithAliases(fields map[FieldName]Field, aliases map[FieldName]FieldName) Schema {
+	internalNameToField := make(map[FieldName]Field)
+	for _, field := range fields {
+		internalNameToField[field.InternalPropertyName] = field
+	}
+	return Schema{
+		Fields:              fields,
+		Aliases:             aliases,
+		internalNameToField: internalNameToField,
+	}
+}
+
+func NewSchema(fields map[FieldName]Field) Schema {
+	return NewSchemaWithAliases(fields, map[FieldName]FieldName{})
+}
+
 func (t FieldName) AsString() string {
 	return string(t)
 }
@@ -26,6 +43,14 @@ func (t TableName) AsString() string {
 	return string(t)
 }
 
+func (s Schema) ResolveFieldByInternalName(fieldName string) (Field, bool) {
+	if field, exists := s.internalNameToField[FieldName(fieldName)]; exists {
+		return field, true
+	} else {
+		return Field{}, false
+	}
+}
+
 func (s Schema) ResolveField(fieldName string) (Field, bool) {
 	if alias, exists := s.Aliases[FieldName(fieldName)]; exists {
 		field, exists := s.Fields[alias]
diff --git a/quesma/schema/schema_test.go b/quesma/schema/schema_test.go
index 7ffacd154..0f5797c22 100644
--- a/quesma/schema/schema_test.go
+++ b/quesma/schema/schema_test.go
@@ -22,44 +22,30 @@ func TestSchema_ResolveField(t *testing.T) {
 			exists:    false,
 		},
 		{
-			name:      "should resolve field",
-			fieldName: "message",
-			schema: Schema{
-				Fields: map[FieldName]Field{
-					"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText},
-				},
-			},
+			name:          "should resolve field",
+			fieldName:     "message",
+			schema:        NewSchema(map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}}),
 			resolvedField: Field{PropertyName: "message", InternalPropertyName: "message", Type: TypeText},
 			exists:        true,
 		},
 		{
-			name:      "should not resolve field",
-			fieldName: "foo",
-			schema: Schema{
-				Fields: map[FieldName]Field{
-					"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText},
-				},
-			},
+			name:          "should not resolve field",
+			fieldName:     "foo",
+			schema:        NewSchema(map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}}),
 			resolvedField: Field{},
 			exists:        false,
 		},
 		{
-			name:      "should resolve aliased field",
-			fieldName: "message_alias",
-			schema: Schema{
-				Fields:  map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}},
-				Aliases: map[FieldName]FieldName{"message_alias": "message"},
-			},
+			name:          "should resolve aliased field",
+			fieldName:     "message_alias",
+			schema:        NewSchemaWithAliases(map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}}, map[FieldName]FieldName{"message_alias": "message"}),
 			resolvedField: Field{PropertyName: "message", InternalPropertyName: "message", Type: TypeText},
 			exists:        true,
 		},
 		{
-			name:      "should not resolve aliased field",
-			fieldName: "message_alias",
-			schema: Schema{
-				Fields:  map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}},
-				Aliases: map[FieldName]FieldName{"message_alias": "foo"},
-			},
+			name:          "should not resolve aliased field",
+			fieldName:     "message_alias",
+			schema:        NewSchemaWithAliases(map[FieldName]Field{"message": {PropertyName: "message", InternalPropertyName: "message", Type: TypeText}}, map[FieldName]FieldName{"message_alias": "foo"}),
 			resolvedField: Field{},
 			exists:        false,
 		},
@@ -76,3 +62,47 @@ func TestSchema_ResolveField(t *testing.T) {
 		})
 	}
 }
+
+func TestSchema_ResolveFieldByInternalName(t *testing.T) {
+	tests := []struct {
+		testName  string
+		schema    Schema
+		fieldName string
+		want      Field
+		found     bool
+	}{
+		{
+			testName:  "empty schema",
+			schema:    NewSchemaWithAliases(map[FieldName]Field{}, map[FieldName]FieldName{}),
+			fieldName: "message",
+			want:      Field{},
+			found:     false,
+		},
+		{
+			testName:  "schema with fields with internal separators, lookup by property name",
+			schema:    NewSchema(map[FieldName]Field{"foo.bar": {PropertyName: "foo.bar", InternalPropertyName: "foo::bar", Type: TypeText}}),
+			fieldName: "foo.bar",
+			want:      Field{},
+			found:     false,
+		},
+		{
+			testName:  "schema with fields with internal separators, lookup by internal name",
+			schema:    NewSchema(map[FieldName]Field{"foo.bar": {PropertyName: "foo.bar", InternalPropertyName: "foo::bar", Type: TypeText}}),
+			fieldName: "foo::bar",
+			want:      Field{PropertyName: "foo.bar", InternalPropertyName: "foo::bar", Type: TypeText},
+			found:     true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.testName, func(t *testing.T) {
+			s := NewSchemaWithAliases(tt.schema.Fields, tt.schema.Aliases)
+			got, found := s.ResolveFieldByInternalName(tt.fieldName)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("ResolveFieldByInternalName() got = %v, want %v", got, tt.want)
+			}
+			if found != tt.found {
+				t.Errorf("ResolveFieldByInternalName() got1 = %v, want %v", found, tt.found)
+			}
+		})
+	}
+}