fix custom schema table filter in DocSchemaInfo

schema "foo" detected "foobar" as its table and run into a
NPE as it couldn't load it

sql/src/main/java/io/crate/metadata/ReferenceInfos.java

@@ -48,8 +48,7 @@
 
 public class ReferenceInfos implements Iterable<SchemaInfo>, ClusterStateListener {
 
-    public static final String SCHEMA_REGEX = "^([^.]+)\\.(.+)";
-    public static final Pattern SCHEMA_PATTERN = Pattern.compile(SCHEMA_REGEX);
+    public static final Pattern SCHEMA_PATTERN = Pattern.compile("^([^.]+)\\.(.+)");
     public static final String DEFAULT_SCHEMA_NAME = "doc";
 
     private final Map<String, SchemaInfo> builtInSchemas;
@@ -150,13 +149,8 @@ public void clusterChanged(ClusterChangedEvent event) {
      * @param name The schema name
      * @return an instance of SchemaInfo for the given name
      */
-    private SchemaInfo getCustomSchemaInfo(final String name) {
-        return new DocSchemaInfo(clusterService, transportPutIndexTemplateAction) {
-            @Override
-            public String name() {
-                return name;
-            }
-        };
+    private SchemaInfo getCustomSchemaInfo(String name) {
+        return new DocSchemaInfo(name, clusterService, transportPutIndexTemplateAction);
     }
 
     /**

sql/src/main/java/io/crate/metadata/doc/DocSchemaInfo.java

@@ -21,11 +21,6 @@
 
 package io.crate.metadata.doc;
 
-import java.util.*;
-import java.util.concurrent.ExecutionException;
-
-import javax.annotation.Nullable;
-
 import com.carrotsearch.hppc.ObjectLookupContainer;
 import com.carrotsearch.hppc.cursors.ObjectCursor;
 import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
@@ -57,53 +52,99 @@
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.inject.Inject;
 
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.*;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Matcher;
+
 public class DocSchemaInfo implements SchemaInfo, ClusterStateListener {
 
     private final ClusterService clusterService;
     private final TransportPutIndexTemplateAction transportPutIndexTemplateAction;
 
-    private final Predicate<String> tablesFilter = new Predicate<String>() {
+    private final static Predicate<String> DOC_SCHEMA_TABLES_FILTER = new Predicate<String>() {
         @Override
         public boolean apply(String input) {
+            //noinspection SimplifiableIfStatement
             if (BlobIndices.isBlobIndex(input)) {
                 return false;
             }
-            if (name().equalsIgnoreCase(ReferenceInfos.DEFAULT_SCHEMA_NAME)) {
-                return !input.matches(ReferenceInfos.SCHEMA_REGEX);
-            } else {
-                return input.startsWith(name());
-            }
+            return !ReferenceInfos.SCHEMA_PATTERN.matcher(input).matches();
         }
     };
 
+    private final Predicate<String> tablesFilter;
+
     private final LoadingCache<String, DocTableInfo> cache = CacheBuilder.newBuilder()
             .maximumSize(10000)
             .build(
                     new CacheLoader<String, DocTableInfo>() {
                         @Override
-                        public DocTableInfo load(String key) throws Exception {
+                        public DocTableInfo load(@Nonnull String key) throws Exception {
                             return innerGetTableInfo(key);
                         }
                     }
             );
+
     private final Function<String, TableInfo> tableInfoFunction;
+    private final String schemaName;
+
 
+    /**
+     * DocSchemaInfo constructor for the default (doc) schema.
+     */
     @Inject
     public DocSchemaInfo(ClusterService clusterService,
                          TransportPutIndexTemplateAction transportPutIndexTemplateAction) {
+        schemaName = ReferenceInfos.DEFAULT_SCHEMA_NAME;
+        this.clusterService = clusterService;
+        clusterService.add(this);
+        this.transportPutIndexTemplateAction = transportPutIndexTemplateAction;
+
+        this.tablesFilter = DOC_SCHEMA_TABLES_FILTER;
+        this.tableInfoFunction = new Function<String, TableInfo>() {
+            @Nullable
+            @Override
+            public TableInfo apply(String input) {
+                return getTableInfo(input);
+            }
+        };
+    }
+
+    /**
+     * constructor used for custom schemas
+     */
+    public DocSchemaInfo(final String schemaName,
+                         ClusterService clusterService,
+                         TransportPutIndexTemplateAction transportPutIndexTemplateAction) {
+        this.schemaName = schemaName;
         this.clusterService = clusterService;
         clusterService.add(this);
         this.transportPutIndexTemplateAction = transportPutIndexTemplateAction;
+
         this.tableInfoFunction = new Function<String, TableInfo>() {
             @Nullable
             @Override
-            public TableInfo apply(@Nullable String input) {
-                if (input.matches(ReferenceInfos.SCHEMA_REGEX)) {
-                    input = input.substring(name().length() + 1);
+            public TableInfo apply(String input) {
+                Matcher matcher = ReferenceInfos.SCHEMA_PATTERN.matcher(input);
+                if (matcher.matches()) {
+                    input = matcher.group(2);
                 }
                 return getTableInfo(input);
             }
         };
+        tablesFilter = new Predicate<String>() {
+            @Override
+            public boolean apply(String input) {
+                //noinspection SimplifiableIfStatement
+                if (BlobIndices.isBlobIndex(input)) {
+                    return false;
+                }
+                Matcher matcher = ReferenceInfos.SCHEMA_PATTERN.matcher(input);
+                return (matcher.matches() && matcher.group(1).equals(schemaName)) ;
+            }
+        };
     }
 
     private DocTableInfo innerGetTableInfo(String name) {
@@ -160,7 +201,7 @@ public Collection<String> tableNames() {
 
     @Override
     public String name() {
-        return ReferenceInfos.DEFAULT_SCHEMA_NAME;
+        return schemaName;
     }
 
     @Override
@@ -241,6 +282,11 @@ private void invalidateAliases(ImmutableOpenMap<String, AliasMetaData> aliases)
         }
     }
 
+    @Override
+    public String toString() {
+        return "DocSchemaInfo(" + name() + ")";
+    }
+
     @Override
     public Iterator<TableInfo> iterator() {
         return Iterators.transform(tableNames().iterator(), tableInfoFunction);

sql/src/test/java/io/crate/integrationtests/CustomSchemaIntegrationTest.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+ * license agreements.  See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership.  Crate licenses
+ * this file to you under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.  You may
+ * obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ *
+ * However, if you have executed another commercial license agreement
+ * with Crate these terms will supersede the license and you may use the
+ * software solely pursuant to the terms of the relevant commercial agreement.
+ */
+
+package io.crate.integrationtests;
+
+import io.crate.test.integration.CrateIntegrationTest;
+import io.crate.testing.TestingHelpers;
+import org.junit.Test;
+
+import static org.hamcrest.Matchers.is;
+
+@CrateIntegrationTest.ClusterScope(scope = CrateIntegrationTest.Scope.GLOBAL)
+public class CustomSchemaIntegrationTest extends SQLTransportIntegrationTest {
+
+
+    @Test
+    public void testInformationSchemaTablesReturnCorrectTablesIfCustomSchemaIsSimilarToTableName() throws Exception {
+        // regression test.. this caused foobar to be detected as a table in the foo schema and caused a NPE
+        execute("create table foobar (id int primary key) with (number_of_replicas = 0)");
+        execute("create table foo.bar (id int primary key) with (number_of_replicas = 0)");
+
+        execute("select schema_name, table_name from information_schema.tables " +
+                "where table_name like 'foo%' or schema_name = 'foo' order by table_name");
+        assertThat(TestingHelpers.printedTable(response.rows()), is("" +
+                "foo| bar\n" +
+                "doc| foobar\n"));
+    }
+}