New IT Framework - InputSource and InputFormat Tests (#13597)

* New IT Framework - InputSource and InputFormat Tests

* Fixing checkstyle errors

* Updating InputSource setup

* Updating queries to use druid DB

* Making metadata setup queries to be idempotent

* Restore intellij files

.travis.yml

@@ -307,10 +307,6 @@ jobs:
       script: *run_integration_test
       after_failure: *integration_test_diags
 
-    - <<: *integration_input_format
-      name: "(Compile=openjdk8, Run=openjdk8) input format integration test with Indexer"
-      env: TESTNG_GROUPS='-Dgroups=input-format' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
-
     - &integration_input_source
       name: "(Compile=openjdk8, Run=openjdk8) input source integration test"
       stage: Tests - phase 2
@@ -320,10 +316,6 @@ jobs:
       script: *run_integration_test
       after_failure: *integration_test_diags
 
-    - <<: *integration_input_source
-      name: "(Compile=openjdk8, Run=openjdk8) input source integration test with Indexer"
-      env: TESTNG_GROUPS='-Dgroups=input-source' JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
-
     - &integration_perfect_rollup_parallel_batch_index
       name: "(Compile=openjdk8, Run=openjdk8) perfect rollup parallel batch index integration test"
       jdk: openjdk8
@@ -549,6 +541,22 @@ jobs:
       env: JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
       script: ./it.sh travis BatchIndex
 
+    - &integration_tests_ex
+      name: "(Compile=openjdk8, Run=openjdk8) input source integration test with Indexer (new)"
+      stage: Tests - phase 2
+      jdk: openjdk8
+      services: *integration_test_services
+      env: JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
+      script: ./it.sh travis InputSource
+
+    - &integration_tests_ex
+      name: "(Compile=openjdk8, Run=openjdk8) input format integration test with Indexer (new)"
+      stage: Tests - phase 2
+      jdk: openjdk8
+      services: *integration_test_services
+      env: JVM_RUNTIME='-Djvm.runtime=8' USE_INDEXER='indexer'
+      script: ./it.sh travis InputFormat
+
     # END - Integration tests for Compile with Java 8 and Run with Java 8
 
     # START - Integration tests for Compile with Java 8 and Run with Java 11

integration-tests-ex/cases/cluster.sh

@@ -79,6 +79,9 @@ function category {
 	# Add an entry here if you create a new category that shares
 	# a definition.
 	case $CATEGORY in
+		"InputSource")
+			export DRUID_INTEGRATION_TEST_GROUP=BatchIndex
+			;;
 		"InputFormat")
 			export DRUID_INTEGRATION_TEST_GROUP=BatchIndex
 			;;

integration-tests-ex/cases/pom.xml

@@ -293,14 +293,23 @@
         </properties>
         </profile>
         <profile>
-          <id>IT-InputFormat</id>
+          <id>IT-InputSource</id>
             <activation>
                 <activeByDefault>false</activeByDefault>
             </activation>
         <properties>
-          <it.category>InputFormat</it.category>
+          <it.category>InputSource</it.category>
         </properties>
         </profile>
+        <profile>
+            <id>IT-InputFormat</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+            </activation>
+            <properties>
+                <it.category>InputFormat</it.category>
+            </properties>
+        </profile>
         <profile>
             <id>IT-AzureDeepStorage</id>
             <activation>

integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/categories/InputFormat.java

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.druid.testsEx.categories;
+
+import org.apache.druid.testsEx.config.Cluster;
+
+@Cluster(BatchIndex.class)
+public class InputFormat
+{
+}

integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/categories/InputSource.java

@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.druid.testsEx.categories;
+
+public class InputSource
+{
+}

integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITHttpInputSourceTest.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.druid.testsEx.indexer;
+
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.InputSource;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.UUID;
+
+@RunWith(DruidTestRunner.class)
+@Category(InputSource.class)
+public class ITHttpInputSourceTest extends AbstractITBatchIndexTest
+{
+  private static final String INDEX_TASK = "/indexer/wikipedia_http_inputsource_task.json";
+  private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_http_inputsource_queries.json";
+
+  @Test
+  public void doTest() throws IOException
+  {
+    final String indexDatasource = "wikipedia_http_inputsource_test_" + UUID.randomUUID();
+    try (final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix())) {
+      doIndexTest(
+          indexDatasource,
+          INDEX_TASK,
+          INDEX_QUERIES_RESOURCE,
+          false,
+          true,
+          true,
+          new Pair<>(false, false)
+      );
+    }
+  }
+}

integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITLocalInputSourceAllInputFormatTest.java

@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.druid.testsEx.indexer;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.tests.indexer.AbstractLocalInputSourceParallelIndexTest;
+import org.apache.druid.testsEx.categories.InputFormat;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+import java.util.Map;
+
+@RunWith(DruidTestRunner.class)
+@Category(InputFormat.class)
+public class ITLocalInputSourceAllInputFormatTest extends AbstractLocalInputSourceParallelIndexTest
+{
+  @Test
+  public void testAvroInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    List fieldList = ImmutableList.of(
+        ImmutableMap.of("name", "timestamp", "type", "string"),
+        ImmutableMap.of("name", "page", "type", "string"),
+        ImmutableMap.of("name", "language", "type", "string"),
+        ImmutableMap.of("name", "user", "type", "string"),
+        ImmutableMap.of("name", "unpatrolled", "type", "string"),
+        ImmutableMap.of("name", "newPage", "type", "string"),
+        ImmutableMap.of("name", "robot", "type", "string"),
+        ImmutableMap.of("name", "anonymous", "type", "string"),
+        ImmutableMap.of("name", "namespace", "type", "string"),
+        ImmutableMap.of("name", "continent", "type", "string"),
+        ImmutableMap.of("name", "country", "type", "string"),
+        ImmutableMap.of("name", "region", "type", "string"),
+        ImmutableMap.of("name", "city", "type", "string"),
+        ImmutableMap.of("name", "added", "type", "int"),
+        ImmutableMap.of("name", "deleted", "type", "int"),
+        ImmutableMap.of("name", "delta", "type", "int")
+    );
+    Map schema = ImmutableMap.of("namespace", "org.apache.druid.data.input",
+                                 "type", "record",
+                                 "name", "wikipedia",
+                                 "fields", fieldList);
+    doIndexTest(InputFormatDetails.AVRO, ImmutableMap.of("schema", schema), new Pair<>(false, false));
+  }
+
+  @Test
+  public void testAvroInputFormatIndexDataIngestionSpecWithoutSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.AVRO, new Pair<>(false, false));
+  }
+
+  @Test
+  public void testJsonInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.JSON, new Pair<>(false, false));
+  }
+
+  @Test
+  public void testTsvInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.TSV, ImmutableMap.of("findColumnsFromHeader", true), new Pair<>(false, false));
+  }
+
+  @Test
+  public void testParquetInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.PARQUET, new Pair<>(false, false));
+  }
+
+  @Test
+  public void testOrcInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.ORC, new Pair<>(false, false));
+  }
+
+  @Test
+  public void testCsvInputFormatIndexDataIngestionSpecWithSchema() throws Exception
+  {
+    doIndexTest(InputFormatDetails.CSV, ImmutableMap.of("findColumnsFromHeader", true), new Pair<>(false, false));
+  }
+}

integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITSqlInputSourceTest.java

@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF 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.
+ */
+
+package org.apache.druid.testsEx.indexer;
+
+import com.google.common.collect.ImmutableList;
+import junitparams.Parameters;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.InputSource;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.UUID;
+import java.util.function.Function;
+
+@RunWith(DruidTestRunner.class)
+@Category(InputSource.class)
+public class ITSqlInputSourceTest extends AbstractITBatchIndexTest
+{
+  private static final String INDEX_TASK = "/indexer/wikipedia_parallel_index_using_sqlinputsource_task.json";
+  private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+
+  public static Object[][] resources()
+  {
+    return new Object[][]{
+        // Multiple query. No filter
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data1", "SELECT * FROM wikipedia_index_data2", "SELECT * FROM wikipedia_index_data3")},
+        // Multiple query. Filter on timestamp column
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data1 WHERE timestamp BETWEEN '2013-08-31 00:00:00' AND '2013-09-02 00:00:00'",
+                          "SELECT * FROM wikipedia_index_data2 WHERE timestamp BETWEEN '2013-08-31 00:00:00' AND '2013-09-02 00:00:00'",
+                          "SELECT * FROM wikipedia_index_data3 WHERE timestamp BETWEEN '2013-09-01 00:00:00' AND '2013-09-02 00:00:00'")},
+        // Multiple query. Filter on data column
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data1 WHERE added > 0",
+                          "SELECT * FROM wikipedia_index_data2 WHERE added > 0",
+                          "SELECT * FROM wikipedia_index_data3 WHERE added > 0")},
+        // Single query. No filter
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data_all")},
+        // Single query. Filter on timestamp column
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data_all WHERE timestamp BETWEEN '2013-08-31 00:00:00' AND '2013-09-02 00:00:00'")},
+        // Single query. Filter on data column
+        {ImmutableList.of("SELECT * FROM wikipedia_index_data_all WHERE added > 0")},
+    };
+  }
+
+  @Test
+  @Parameters(method = "resources")
+  public void testIndexData(List<String> sqlQueries) throws Exception
+  {
+    final String indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+    try (
+        final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+    ) {
+      final Function<String, String> sqlInputSourcePropsTransform = spec -> {
+        try {
+          spec = StringUtils.replace(
+              spec,
+              "%%PARTITIONS_SPEC%%",
+              jsonMapper.writeValueAsString(new DynamicPartitionsSpec(null, null))
+          );
+          return StringUtils.replace(
+              spec,
+              "%%SQL_QUERY%%",
+              jsonMapper.writeValueAsString(sqlQueries)
+          );
+        }
+        catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+      };
+
+      doIndexTest(
+          indexDatasource,
+          INDEX_TASK,
+          sqlInputSourcePropsTransform,
+          INDEX_QUERIES_RESOURCE,
+          false,
+          true,
+          true,
+          new Pair<>(false, false)
+      );
+    }
+
+  }
+}

integration-tests-ex/cases/src/test/resources/cluster/InputSource/docker.yaml

@@ -0,0 +1,328 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF 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.
+#-------------------------------------------------------------------------
+
+# Definition of the input source test cluster.
+# See https://yaml.org/spec/1.2.2 for more about YAML
+include:
+  - /cluster/Common/zk-metastore.yaml
+
+druid:
+  coordinator:
+    instances:
+      - port: 8081
+  overlord:
+    instances:
+      - port: 8090
+  broker:
+    instances:
+      - port: 8082
+  router:
+    instances:
+      - port: 8888
+  historical:
+    instances:
+      - port: 8083
+  indexer:
+    instances:
+      - port: 8091
+
+# Metastore initialization queries.
+# REPLACE is used so that the statements are idempotent
+# The fancy formatting is for human consumption, it is compacted internally
+metastoreInit:
+  - sql: |
+      create table if not exists druid.wikipedia_index_data1(
+          timestamp VARCHAR(100) NOT NULL,
+          page VARCHAR(100),
+          language VARCHAR(40),
+          user VARCHAR(100),
+          unpatrolled VARCHAR(100),
+          newPage VARCHAR(40),
+          robot VARCHAR(100),
+          anonymous VARCHAR(100),
+          namespace VARCHAR(40),
+          continent VARCHAR(100),
+          country VARCHAR(100),
+          region VARCHAR(40),
+          city VARCHAR(100),
+          added INT,
+          deleted INT,
+          delta INT,
+          UNIQUE KEY `unique_index` (`timestamp`,`page`, `user`)
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data1 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T01:02:33Z', 'Gypsy Danger', 'en', 'nuclear',
+        'true', 'true', 'false', 'false', 'article', 'North America',
+        'United States', 'Bay Area', 'San Francisco', 57, 200, -143
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data1 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T03:32:45Z', 'Striker Eureka', 'en', 'speed', 
+        'false', 'true', 'true', 'false', 'wikipedia', 'Australia', 
+        'Australia', 'Cantebury', 'Syndey', 459, 129, 330
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data1 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T07:11:21Z', 'Cherno Alpha', 'ru', 'masterYi', 
+        'false', 'true', 'true', 'false', 'article', 'Asia', 
+        'Russia', 'Oblast', 'Moscow', 123, 12, 111
+      );
+
+  - sql: |
+      create table if not exists druid.wikipedia_index_data2(
+          timestamp VARCHAR(100) NOT NULL,
+          page VARCHAR(100),
+          language VARCHAR(40),
+          user VARCHAR(100),
+          unpatrolled VARCHAR(100),
+          newPage VARCHAR(40),
+          robot VARCHAR(100),
+          anonymous VARCHAR(100),
+          namespace VARCHAR(40),
+          continent VARCHAR(100),
+          country VARCHAR(100),
+          region VARCHAR(40),
+          city VARCHAR(100),
+          added INT,
+          deleted INT,
+          delta INT,
+          UNIQUE KEY `unique_index` (`timestamp`,`page`, `user`)
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data2 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T11:58:39Z', 'Crimson Typhoon', 'zh', 'triplets', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'China', 'Shanxi', 'Taiyuan', 905, 5, 900
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data2 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T12:41:27Z', 'Coyote Tango', 'ja', 'stringer', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'Japan', 'Kanto', 'Tokyo', 1, 10, -9
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data2 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T01:02:33Z', 'Gypsy Danger', 'en', 'nuclear', 
+        'true', 'true', 'false', 'false', 'article', 'North America', 
+        'United States', 'Bay Area', 'San Francisco', 57, 200, -143
+      );
+
+  - sql: |
+      create table if not exists druid.wikipedia_index_data3(
+          timestamp VARCHAR(100) NOT NULL,
+          page VARCHAR(100),
+          language VARCHAR(40),
+          user VARCHAR(100),
+          unpatrolled VARCHAR(100),
+          newPage VARCHAR(40),
+          robot VARCHAR(100),
+          anonymous VARCHAR(100),
+          namespace VARCHAR(40),
+          continent VARCHAR(100),
+          country VARCHAR(100),
+          region VARCHAR(40),
+          city VARCHAR(100),
+          added INT,
+          deleted INT,
+          delta INT,
+          UNIQUE KEY `unique_index` (`timestamp`,`page`, `user`)
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data3 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T03:32:45Z', 'Striker Eureka', 'en', 'speed', 
+        'false', 'true', 'true', 'false', 'wikipedia', 'Australia', 
+        'Australia', 'Cantebury', 'Syndey', 459, 129, 330
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data3 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T07:11:21Z', 'Cherno Alpha', 'ru', 'masterYi', 
+        'false', 'true', 'true', 'false', 'article', 'Asia', 
+        'Russia', 'Oblast', 'Moscow', 123, 12, 111
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data3 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T11:58:39Z', 'Crimson Typhoon', 'zh', 'triplets', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'China', 'Shanxi', 'Taiyuan', 905, 5, 900
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data3 (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T12:41:27Z', 'Coyote Tango', 'ja', 'stringer', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'Japan', 'Kanto', 'Tokyo', 1, 10, -9
+      );
+
+  - sql: |
+      create table if not exists druid.wikipedia_index_data_all(
+          timestamp VARCHAR(100) NOT NULL,
+          page VARCHAR(100),
+          language VARCHAR(40),
+          user VARCHAR(100),
+          unpatrolled VARCHAR(100),
+          newPage VARCHAR(40),
+          robot VARCHAR(100),
+          anonymous VARCHAR(100),
+          namespace VARCHAR(40),
+          continent VARCHAR(100),
+          country VARCHAR(100),
+          region VARCHAR(40),
+          city VARCHAR(100),
+          added INT,
+          deleted INT,
+          delta INT,
+          UNIQUE KEY `unique_index` (`timestamp`,`page`, `user`)
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T01:02:33Z', 'Gypsy Danger', 'en', 'nuclear',
+        'true', 'true', 'false', 'false', 'article', 'North America',
+        'United States', 'Bay Area', 'San Francisco', 57, 200, -143
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T03:32:45Z', 'Striker Eureka', 'en', 'speed', 
+        'false', 'true', 'true', 'false', 'wikipedia', 'Australia', 
+        'Australia', 'Cantebury', 'Syndey', 459, 129, 330
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T07:11:21Z', 'Cherno Alpha', 'ru', 'masterYi', 
+        'false', 'true', 'true', 'false', 'article', 'Asia', 
+        'Russia', 'Oblast', 'Moscow', 123, 12, 111
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T11:58:39Z', 'Crimson Typhoon', 'zh', 'triplets', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'China', 'Shanxi', 'Taiyuan', 905, 5, 900
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-08-31T12:41:27Z', 'Coyote Tango', 'ja', 'stringer', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'Japan', 'Kanto', 'Tokyo', 1, 10, -9
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T01:02:33Z', 'Gypsy Danger', 'en', 'nuclear', 
+        'true', 'true', 'false', 'false', 'article', 'North America', 
+        'United States', 'Bay Area', 'San Francisco', 57, 200, -143
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T03:32:45Z', 'Striker Eureka', 'en', 'speed', 
+        'false', 'true', 'true', 'false', 'wikipedia', 'Australia', 
+        'Australia', 'Cantebury', 'Syndey', 459, 129, 330
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T07:11:21Z', 'Cherno Alpha', 'ru', 'masterYi', 
+        'false', 'true', 'true', 'false', 'article', 'Asia', 
+        'Russia', 'Oblast', 'Moscow', 123, 12, 111
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T11:58:39Z', 'Crimson Typhoon', 'zh', 'triplets', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'China', 'Shanxi', 'Taiyuan', 905, 5, 900
+      );
+
+  - sql: |
+      REPLACE INTO druid.wikipedia_index_data_all (
+        timestamp, page, language, user, unpatrolled, newPage, robot, anonymous, namespace, continent, country, region, city, added, deleted, delta
+      )
+      VALUES (
+        '2013-09-01T12:41:27Z', 'Coyote Tango', 'ja', 'stringer', 
+        'true', 'false', 'true', 'false', 'wikipedia', 'Asia', 
+        'Japan', 'Kanto', 'Tokyo', 1, 10, -9
+      );

integration-tests-ex/cases/src/test/resources/indexer/wikipedia_parallel_index_using_sqlinputsource_task.json

@@ -71,7 +71,7 @@
         "database": {
           "type": "mysql",
           "connectorConfig": {
-            "connectURI": "jdbc:mysql://druid-metadata-storage/sqlinputsource",
+            "connectURI": "jdbc:mysql://metadata/druid",
             "user": "druid",
             "password": "diurd"
           }