You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2022/12/09 05:42:45 UTC
[GitHub] [druid] abhagraw opened a new pull request, #13535: MSQ s3 deep storage tests
abhagraw opened a new pull request, #13535:
URL: https://github.com/apache/druid/pull/13535
Adds IT for SQL Based ingestion from S3 deep storage.
Migrates existing s3DeepStorage tests to the new IT framework.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1051934449
##########
integration-tests-ex/cases/cluster/S3DeepStorage/docker-compose.yaml:
##########
@@ -0,0 +1,121 @@
+# 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.
+
+networks:
+ druid-it-net:
+ name: druid-it-net
+ ipam:
+ config:
+ - subnet: 172.172.172.0/24
+
+services:
+ zookeeper:
+ extends:
+ file: ../Common/dependencies.yaml
+ service: zookeeper
+
+ metadata:
+ extends:
+ file: ../Common/dependencies.yaml
+ service: metadata
+
+ coordinator:
+ extends:
+ file: ../Common/druid.yaml
+ service: coordinator
+ container_name: coordinator
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ # The frequency with which the coordinator polls the database
+ # for changes. The DB population code has to wait at least this
+ # long for the coordinator to notice changes.
+ - druid_manager_segments_pollDuration=PT5S
+ - druid_coordinator_period=PT10S
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+ - metadata
+
+ overlord:
+ extends:
+ file: ../Common/druid.yaml
+ service: overlord
+ container_name: overlord
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+ - metadata
+
+ broker:
+ extends:
+ file: ../Common/druid.yaml
+ service: broker
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ router:
+ extends:
+ file: ../Common/druid.yaml
+ service: router
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ historical:
+ extends:
+ file: ../Common/druid.yaml
+ service: historical
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ indexer:
+ extends:
+ file: ../Common/druid.yaml
+ service: indexer
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - druid_test_loadList=druid-s3-extensions
+ - druid_storage_type=s3
+ - druid_storage_bucket=${DRUID_CLOUD_BUCKET}
+ # Using DRUID_CLOUD_PATH env as baseKey as well.
+ - druid_storage_baseKey=${DRUID_CLOUD_PATH}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
Review Comment:
It is already documented in this file - https://github.com/apache/druid/blob/master/integration-tests-ex/docs/test-config.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on PR #13535:
URL: https://github.com/apache/druid/pull/13535#issuecomment-1354748832
@paul-rogers Have merged `AbstractITSQLBasedIngestionTest` with `AbstractITBatchIndexTest` and moved all the s3 related code to `AbstractS3InputSourceParallelIndexTest`. This eliminates most of the code duplication.
Now msq test and native batch ingestion tests will inherit from the same class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] github-code-scanning[bot] commented on a diff in pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
github-code-scanning[bot] commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1055370005
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -93,14 +99,149 @@
private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
Review Comment:
## Field masks field in super class
This field shadows another field called [queryHelper](1) in a superclass.
[Show more details](https://github.com/apache/druid/security/code-scanning/3575)
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITAzureSQLBasedIngestionTest.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractAzureInputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment -
+ * <p>
+ * The AZURE account, key and container should be set in AZURE_ACCOUNT, AZURE_KEY and AZURE_CONTAINER respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/azure.html">Azure Deep Storage setup in druid</a>
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITAzureSQLBasedIngestionTest extends AbstractAzureInputSourceParallelIndexTest
+{
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
Review Comment:
## Field masks field in super class
This field shadows another field called [jsonMapper](1) in a superclass.
[Show more details](https://github.com/apache/druid/security/code-scanning/3577)
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITGcsSQLBasedIngestionTest.java:
##########
@@ -0,0 +1,55 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.GcsDeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractGcsInputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+@RunWith(DruidTestRunner.class)
+@Category(GcsDeepStorage.class)
+public class ITGcsSQLBasedIngestionTest extends AbstractGcsInputSourceParallelIndexTest
+{
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
Review Comment:
## Field masks field in super class
This field shadows another field called [jsonMapper](1) in a superclass.
[Show more details](https://github.com/apache/druid/security/code-scanning/3576)
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractS3InputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 Bucket to store in (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in
+ * AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractS3InputSourceParallelIndexTest
+{
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
Review Comment:
## Field masks field in super class
This field shadows another field called [jsonMapper](1) in a superclass.
[Show more details](https://github.com/apache/druid/security/code-scanning/3578)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1065334297
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
Review Comment:
It varies inside the test function. See line 138.
Do you want me to use a constant here and use a different variable there?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1065335424
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
+
+ public static Object[][] resources()
Review Comment:
Java does not allow me directly create an array of this type, since it is strongly typed.
So I am creating it using Object[][].
I think this is why it was previously done using the same way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
cryptoe commented on PR #13535:
URL: https://github.com/apache/druid/pull/13535#issuecomment-1378220308
Thanks, @abhagraw for the contribution. This really helps out moving off the old framework!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe merged pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
cryptoe merged PR #13535:
URL: https://github.com/apache/druid/pull/13535
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
paul-rogers commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1047483660
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITS3ToS3ParallelIndexTest.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 junitparams.Parameters;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+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;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must:
+ * 1) Set the bucket and path for your data. This can be done by setting -Ddruid.test.config.cloudBucket and
+ * -Ddruid.test.config.cloudPath or setting "cloud_bucket" and "cloud_path" in the config file.
+ * 2) Copy wikipedia_index_data1.json, wikipedia_index_data2.json, and wikipedia_index_data3.json
+ * located in integration-tests/src/test/resources/data/batch_index/json to your S3 at the location set in step 1.
+ * 3) Provide -Doverride.config.path=<PATH_TO_FILE> with s3 credentials/configs set. See
Review Comment:
I'm not sure if the new ITs implement this feature. If it did, we have no good way in the new ITs to pass this information into the test, as we use the same Java command line for all tests. In the new IT framework, we prefer to use env vars, not command line options.
I wonder, are all these comments just copy/pasted from old tests without adjusting them for the new IT framework? If so, we're doing future readers a dis-service by providing incorrect info. If we don't have time to provide correct comments, just remove the comments and let the developer rediscover how things work without being sent down the wrong path.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITS3ToS3ParallelIndexTest.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 junitparams.Parameters;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+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;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must:
+ * 1) Set the bucket and path for your data. This can be done by setting -Ddruid.test.config.cloudBucket and
Review Comment:
Is this comment accurate? We discussed setting the env var directly. For the above to work, something would have to pass the flags to the test process, but the new ITs are not set up to do that.
##########
integration-tests-ex/cases/cluster/S3DeepStorage/docker-compose.yaml:
##########
@@ -0,0 +1,121 @@
+# 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.
+
+networks:
+ druid-it-net:
+ name: druid-it-net
+ ipam:
+ config:
+ - subnet: 172.172.172.0/24
+
+services:
+ zookeeper:
+ extends:
+ file: ../Common/dependencies.yaml
+ service: zookeeper
+
+ metadata:
+ extends:
+ file: ../Common/dependencies.yaml
+ service: metadata
+
+ coordinator:
+ extends:
+ file: ../Common/druid.yaml
+ service: coordinator
+ container_name: coordinator
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ # The frequency with which the coordinator polls the database
+ # for changes. The DB population code has to wait at least this
+ # long for the coordinator to notice changes.
+ - druid_manager_segments_pollDuration=PT5S
+ - druid_coordinator_period=PT10S
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+ - metadata
+
+ overlord:
+ extends:
+ file: ../Common/druid.yaml
+ service: overlord
+ container_name: overlord
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+ - metadata
+
+ broker:
+ extends:
+ file: ../Common/druid.yaml
+ service: broker
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ router:
+ extends:
+ file: ../Common/druid.yaml
+ service: router
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ historical:
+ extends:
+ file: ../Common/druid.yaml
+ service: historical
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - AWS_REGION=${AWS_REGION}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
+ depends_on:
+ - zookeeper
+
+ indexer:
+ extends:
+ file: ../Common/druid.yaml
+ service: indexer
+ environment:
+ - DRUID_INTEGRATION_TEST_GROUP=${DRUID_INTEGRATION_TEST_GROUP}
+ - druid_test_loadList=druid-s3-extensions
+ - druid_storage_type=s3
+ - druid_storage_bucket=${DRUID_CLOUD_BUCKET}
+ # Using DRUID_CLOUD_PATH env as baseKey as well.
+ - druid_storage_baseKey=${DRUID_CLOUD_PATH}
+ - druid_s3_accessKey=${AWS_ACCESS_KEY_ID}
+ - druid_s3_secretKey=${AWS_SECRET_ACCESS_KEY}
Review Comment:
Let's document the required env vars somewhere. Perhaps we add a Markdown page that lists the requirements for each test. Or, we document these at the top of the Java file. Be sure to list the env vars used by the server vs. those used by the test.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITS3ToS3ParallelIndexTest.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 junitparams.Parameters;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+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;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must:
+ * 1) Set the bucket and path for your data. This can be done by setting -Ddruid.test.config.cloudBucket and
+ * -Ddruid.test.config.cloudPath or setting "cloud_bucket" and "cloud_path" in the config file.
+ * 2) Copy wikipedia_index_data1.json, wikipedia_index_data2.json, and wikipedia_index_data3.json
+ * located in integration-tests/src/test/resources/data/batch_index/json to your S3 at the location set in step 1.
Review Comment:
A comment somewhere said that the test itself is doing this step. If this test were run as part of a build, we'd have to insert this step into the script which launches the test. Did we do so?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
Review Comment:
A common error will be to leave one or both of these unset. Even though this is a test, and we want to go fast, future users will appreciate it if we check if these values are set and give a good error if not set.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
Review Comment:
Again, check the value and given an error.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
Review Comment:
This seems a clone of the `resources` in `AbstractS3InputSourceParallelIndexTest`. Even though we want to move fast, it is still helpful to follow good practices and factor out common code.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
+ .build();
+ }
+
+ public static String[] fileList()
+ {
+ return new String[] {
+ WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3
+ };
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ s3Client.putObject(
+ System.getenv("DRUID_CLOUD_BUCKET"),
+ System.getenv("DRUID_CLOUD_PATH") + "/" + file,
+ new File(localPath + file)
+ );
+ }
+ }
+
+ @AfterClass
+ public static void deleteFilesFromS3()
+ {
+ // Delete uploaded data files
+ DeleteObjectsRequest delObjReq = new DeleteObjectsRequest(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withKeys(fileList());
+ s3Client.deleteObjects(delObjReq);
+
+ // Delete segments created by druid
+ ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
+ .withBucketName(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withPrefix(System.getenv("DRUID_CLOUD_PATH") + "/" + datasource + "/");
Review Comment:
See comments above.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
Review Comment:
Some explanation would be helpful. I see code to upload and delete files. There seems to be a step to run an MSQ query. Not sure if there is a step to validate the results. Please add a note to explain what this test does. I, as a reviewer, don't want to have to reverse-engineer the intent from the code.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
+ .build();
+ }
+
+ public static String[] fileList()
+ {
+ return new String[] {
+ WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3
+ };
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ s3Client.putObject(
+ System.getenv("DRUID_CLOUD_BUCKET"),
+ System.getenv("DRUID_CLOUD_PATH") + "/" + file,
+ new File(localPath + file)
+ );
+ }
+ }
+
+ @AfterClass
+ public static void deleteFilesFromS3()
+ {
+ // Delete uploaded data files
+ DeleteObjectsRequest delObjReq = new DeleteObjectsRequest(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withKeys(fileList());
+ s3Client.deleteObjects(delObjReq);
+
+ // Delete segments created by druid
+ ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
+ .withBucketName(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withPrefix(System.getenv("DRUID_CLOUD_PATH") + "/" + datasource + "/");
+
+ ObjectListing objectListing = s3Client.listObjects(listObjectsRequest);
+
+ while (true) {
+ for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries()) {
+ s3Client.deleteObject(System.getenv("DRUID_CLOUD_BUCKET"), objectSummary.getKey());
+ }
+ if (objectListing.isTruncated()) {
+ objectListing = s3Client.listNextBatchOfObjects(objectListing);
+ } else {
+ break;
+ }
+ }
+ }
+
+ @Test
+ @Parameters(method = "test_cases")
+ @TestCaseName("Test_{index} ({0})")
+ public void testSQLBasedBatchIngestion(Pair<String, List> s3InputSource)
+ {
+ try {
+ String sqlTask = getStringFromFileAndReplaceDatasource(CLOUD_INGEST_SQL, datasource);
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ Map<String, Object> context = ImmutableMap.of("finalizeAggregations", false,
+ "maxNumTasks", 5,
+ "groupByEnableMultiValueUnnesting", false);
+
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+
+ // Setting the correct object path in the sqlTask.
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%BUCKET%%",
+ config.getCloudBucket() // Getting from DRUID_CLOUD_BUCKET env variable
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%PATH%%",
+ config.getCloudPath() // Getting from DRUID_CLOUD_PATH env variable
+ );
+
+ submitTask(sqlTask, datasource, context);
+ doTestQuery(INDEX_QUERIES_FILE, datasource);
Review Comment:
Does this verify the results by running queries against the uploaded data? If so, please add a comment to note this.
If this does not verify results, then we need to add a step to do so. Else, this is not much of a test since a simple direct return from `submitTask()` (while doing nothing) would count as a success.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
+ .build();
+ }
+
+ public static String[] fileList()
+ {
+ return new String[] {
+ WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3
+ };
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
Review Comment:
This function has no try-catch. If one upload fails, we don't delete the files already uploaded. This will cause the next test to fail. It is helpful to not only get code done quickly, but to also get it done well to prevent cascading errors.
So, please provide try/catch blocks that ensure that files are cleaned up even if something fails.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
+ .build();
+ }
+
+ public static String[] fileList()
+ {
+ return new String[] {
+ WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3
+ };
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ s3Client.putObject(
+ System.getenv("DRUID_CLOUD_BUCKET"),
+ System.getenv("DRUID_CLOUD_PATH") + "/" + file,
Review Comment:
Please check that the env var is set. This code appears to be a quick hack to get things done ASAP, but without normal coding practices. This code bypasses the mechanism that exists for this Druid-specific value. It does not check if the value is set. If fetches the env var for every object rather than fetching once and reusing the value (which is what the existing code does.)
I'm hesitant to approve code that is overly quick and dirty.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractITSQLBasedIngestion
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static AmazonS3 s3Client = s3Client();
+ private static String datasource = "wikipedia_cloud_index_msq";
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] test_cases()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static AmazonS3 s3Client()
+ {
+ AWSCredentials credentials = new BasicAWSCredentials(
+ System.getenv("AWS_ACCESS_KEY_ID"),
+ System.getenv("AWS_SECRET_ACCESS_KEY")
+ );
+ return AmazonS3ClientBuilder
+ .standard()
+ .withCredentials(new AWSStaticCredentialsProvider(credentials))
+ .withRegion(System.getenv("AWS_REGION"))
+ .build();
+ }
+
+ public static String[] fileList()
+ {
+ return new String[] {
+ WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3
+ };
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ s3Client.putObject(
+ System.getenv("DRUID_CLOUD_BUCKET"),
+ System.getenv("DRUID_CLOUD_PATH") + "/" + file,
+ new File(localPath + file)
+ );
+ }
+ }
+
+ @AfterClass
+ public static void deleteFilesFromS3()
+ {
+ // Delete uploaded data files
+ DeleteObjectsRequest delObjReq = new DeleteObjectsRequest(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withKeys(fileList());
+ s3Client.deleteObjects(delObjReq);
+
+ // Delete segments created by druid
+ ListObjectsRequest listObjectsRequest = new ListObjectsRequest()
+ .withBucketName(System.getenv("DRUID_CLOUD_BUCKET"))
+ .withPrefix(System.getenv("DRUID_CLOUD_PATH") + "/" + datasource + "/");
+
+ ObjectListing objectListing = s3Client.listObjects(listObjectsRequest);
+
+ while (true) {
+ for (S3ObjectSummary objectSummary : objectListing.getObjectSummaries()) {
+ s3Client.deleteObject(System.getenv("DRUID_CLOUD_BUCKET"), objectSummary.getKey());
+ }
+ if (objectListing.isTruncated()) {
+ objectListing = s3Client.listNextBatchOfObjects(objectListing);
+ } else {
+ break;
+ }
+ }
+ }
+
+ @Test
+ @Parameters(method = "test_cases")
+ @TestCaseName("Test_{index} ({0})")
+ public void testSQLBasedBatchIngestion(Pair<String, List> s3InputSource)
+ {
+ try {
+ String sqlTask = getStringFromFileAndReplaceDatasource(CLOUD_INGEST_SQL, datasource);
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ Map<String, Object> context = ImmutableMap.of("finalizeAggregations", false,
+ "maxNumTasks", 5,
+ "groupByEnableMultiValueUnnesting", false);
+
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+
+ // Setting the correct object path in the sqlTask.
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%BUCKET%%",
+ config.getCloudBucket() // Getting from DRUID_CLOUD_BUCKET env variable
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%PATH%%",
+ config.getCloudPath() // Getting from DRUID_CLOUD_PATH env variable
+ );
+
+ submitTask(sqlTask, datasource, context);
Review Comment:
Does this submit the MSQ ingestion task? If so, please add a comment to note this.
For MSQ, we have to submit a task to a specific endpoint, poll to wait for completion, and check completion status. Do we do that somewhere?
##########
integration-tests-ex/cases/src/test/resources/multi-stage-query/wikipedia_cloud_index_msq.sql:
##########
@@ -0,0 +1,32 @@
+REPLACE INTO "%%DATASOURCE%%" OVERWRITE ALL
+WITH "source" as (SELECT * FROM TABLE(
+ EXTERN(
+ '{"type":"s3","%%INPUT_SOURCE_PROPERTY_KEY%%":%%INPUT_SOURCE_PROPERTY_VALUE%%}',
+ '{"type":"json"}',
+ '[{"name":"timestamp","type":"string"},{"name":"page","type":"string"},{"name":"language","type":"string"},{"name":"user","type":"string"},{"name":"unpatrolled","type":"string"},{"name":"newPage","type":"string"},{"name":"robot","type":"string"},{"name":"anonymous","type":"string"},{"name":"namespace","type":"string"},{"name":"continent","type":"string"},{"name":"country","type":"string"},{"name":"region","type":"string"},{"name":"city","type":"string"},{"name":"added","type":"double"},{"name":"deleted","type":"double"},{"name":"delta","type":"double"}]'
+ )
+))
+SELECT
+ TIME_FLOOR(CASE WHEN CAST("timestamp" AS BIGINT) > 0 THEN MILLIS_TO_TIMESTAMP(CAST("timestamp" AS BIGINT)) ELSE TIME_PARSE("timestamp") END, 'PT1S') AS __time,
+ "page",
+ "language",
+ "user",
+ "unpatrolled",
+ "newPage",
+ "robot",
+ "anonymous",
+ "namespace",
+ "continent",
+ "country",
+ "region",
+ "city",
+ COUNT(*) AS "count",
+ SUM("added") AS "added",
+ SUM("deleted") AS "deleted",
+ SUM("delta") AS "delta",
+ APPROX_COUNT_DISTINCT_DS_THETA("user") AS "thetaSketch",
+ DS_QUANTILES_SKETCH("delta") AS "quantilesDoublesSketch",
+ APPROX_COUNT_DISTINCT_DS_HLL("user") AS "HLLSketchBuild"
+FROM "source"
+GROUP BY 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13
+PARTITIONED BY DAY
Review Comment:
Missing newline
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,206 @@
+/*
+ * 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.msq;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET, DRUID_CLOUD_PATH, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
Review Comment:
Please explain what these should be set to.
`DRUID_CLOUD_BUCKET`: The full URL or just a name?
`DRUID_CLOUD_PATH`: Same question. What is the relationship between this item and the above.
`AWS_*` these are probably standard AWS settings?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1052937173
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
Review Comment:
the putObject method that is used to upload files - does take File object.
I am creating the file object in S3TestUtil after receiving the strings there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] cryptoe commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
cryptoe commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1052170067
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -91,16 +97,149 @@ public String getFolderSuffix()
}
}
- private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
+ public static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
Review Comment:
This should be private. You can use a new logger in the sub-class if that is required.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/S3TestUtil.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.testing.utils;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class S3TestUtil
Review Comment:
I think there might be a better way to do this. Maybe use `StorageConnector` interface as you would have to do is for every type of sink ie (GCS, S3, Azure, Local)
cc @paul-rogers wdyt ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
paul-rogers commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1052462276
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -91,16 +97,149 @@ public String getFolderSuffix()
}
}
- private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
+ public static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Rule
+ public TestWatcher watchman = new TestWatcher()
+ {
+ @Override
+ public void starting(Description d)
+ {
+ LOG.info("RUNNING %s", d.getDisplayName());
+ }
+
+ @Override
+ public void failed(Throwable e, Description d)
+ {
+ LOG.error("FAILED %s", d.getDisplayName());
+ }
+
+ @Override
+ public void finished(Description d)
+ {
+ LOG.info("FINISHED %s", d.getDisplayName());
+ }
+ };
+
+ /**
+ * Reads file as utf-8 string and replace %%DATASOURCE%% with the provide datasource value.
+ */
+ protected String getStringFromFileAndReplaceDatasource(String filePath, String datasource)
+ {
+ String fileString;
+ try {
+ InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(filePath);
+ fileString = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ catch (IOException e) {
+ throw new ISE(e, "could not read query file: %s", filePath);
+ }
+
+ fileString = StringUtils.replace(
+ fileString,
+ "%%DATASOURCE%%",
+ datasource
+ );
+
+ return fileString;
+ }
+
+ /**
+ * Reads native queries from a file and runs against the provided datasource.
+ */
+ protected void doTestQuery(String dataSource, String queryFilePath)
+ {
+ try {
+ String query = getStringFromFileAndReplaceDatasource(queryFilePath, dataSource);
+ queryHelper.testQueriesFromString(query);
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while running test query at path " + queryFilePath);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTask(String sqlTask, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("SqlTask - \n %s", sqlTask);
+
+ // Submit the tasks and wait for the datasource to get loaded
+ msqHelper.submitMsqTaskAndWaitForCompletion(
+ sqlTask,
+ msqContext
+ );
+
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTaskFromFile(String sqlFilePath, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ String sqlTask = getStringFromFileAndReplaceDatasource(sqlFilePath, datasource);
+ submitMSQTask(sqlTask, datasource, msqContext);
+ }
+
+ /**
+ * Runs a SQL ingest test.
+ *
+ * @param sqlFilePath path of file containing the sql query.
+ * @param queryFilePath path of file containing the native test queries to be run on the ingested datasource.
+ * @param datasource name of the datasource. %%DATASOURCE%% in the sql and queries will be replaced with this value.
+ * @param msqContext context parameters to be passed with MSQ API call.
+ */
+ protected void runMSQTaskandTestQueries(String sqlFilePath,
+ String queryFilePath,
+ String datasource,
+ Map<String, Object> msqContext) throws Exception
Review Comment:
Nit: I think Druid prefers the following formatting:
```java
protected void runMSQTaskandTestQueries(
String sqlFilePath,
String queryFilePath,
String datasource,
Map<String, Object> msqContext
) throws Exception
```
At least, I always get comments if I forget to format things the above way. The commenter notes "we haven't figured out how to get Checkstyle to enforce this style, but it what we want."
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -91,16 +97,149 @@ public String getFolderSuffix()
}
}
- private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
+ public static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Rule
+ public TestWatcher watchman = new TestWatcher()
+ {
+ @Override
+ public void starting(Description d)
+ {
+ LOG.info("RUNNING %s", d.getDisplayName());
+ }
+
+ @Override
+ public void failed(Throwable e, Description d)
+ {
+ LOG.error("FAILED %s", d.getDisplayName());
+ }
+
+ @Override
+ public void finished(Description d)
+ {
+ LOG.info("FINISHED %s", d.getDisplayName());
+ }
+ };
+
+ /**
+ * Reads file as utf-8 string and replace %%DATASOURCE%% with the provide datasource value.
+ */
+ protected String getStringFromFileAndReplaceDatasource(String filePath, String datasource)
+ {
+ String fileString;
+ try {
+ InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(filePath);
+ fileString = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ catch (IOException e) {
+ throw new ISE(e, "could not read query file: %s", filePath);
+ }
+
+ fileString = StringUtils.replace(
+ fileString,
+ "%%DATASOURCE%%",
+ datasource
+ );
+
+ return fileString;
+ }
+
+ /**
+ * Reads native queries from a file and runs against the provided datasource.
+ */
+ protected void doTestQuery(String dataSource, String queryFilePath)
+ {
+ try {
+ String query = getStringFromFileAndReplaceDatasource(queryFilePath, dataSource);
+ queryHelper.testQueriesFromString(query);
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while running test query at path " + queryFilePath);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTask(String sqlTask, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("SqlTask - \n %s", sqlTask);
+
+ // Submit the tasks and wait for the datasource to get loaded
+ msqHelper.submitMsqTaskAndWaitForCompletion(
+ sqlTask,
+ msqContext
+ );
+
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
Review Comment:
Sumits -> Submits
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -91,16 +97,149 @@ public String getFolderSuffix()
}
}
- private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
+ public static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Rule
+ public TestWatcher watchman = new TestWatcher()
+ {
+ @Override
+ public void starting(Description d)
+ {
+ LOG.info("RUNNING %s", d.getDisplayName());
+ }
+
+ @Override
+ public void failed(Throwable e, Description d)
+ {
+ LOG.error("FAILED %s", d.getDisplayName());
+ }
+
+ @Override
+ public void finished(Description d)
+ {
+ LOG.info("FINISHED %s", d.getDisplayName());
+ }
+ };
+
+ /**
+ * Reads file as utf-8 string and replace %%DATASOURCE%% with the provide datasource value.
+ */
+ protected String getStringFromFileAndReplaceDatasource(String filePath, String datasource)
+ {
+ String fileString;
+ try {
+ InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(filePath);
+ fileString = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ catch (IOException e) {
+ throw new ISE(e, "could not read query file: %s", filePath);
+ }
+
+ fileString = StringUtils.replace(
+ fileString,
+ "%%DATASOURCE%%",
+ datasource
+ );
+
+ return fileString;
+ }
+
+ /**
+ * Reads native queries from a file and runs against the provided datasource.
+ */
+ protected void doTestQuery(String dataSource, String queryFilePath)
+ {
+ try {
+ String query = getStringFromFileAndReplaceDatasource(queryFilePath, dataSource);
+ queryHelper.testQueriesFromString(query);
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while running test query at path " + queryFilePath);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTask(String sqlTask, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("SqlTask - \n %s", sqlTask);
+
+ // Submit the tasks and wait for the datasource to get loaded
+ msqHelper.submitMsqTaskAndWaitForCompletion(
+ sqlTask,
+ msqContext
+ );
+
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTaskFromFile(String sqlFilePath, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ String sqlTask = getStringFromFileAndReplaceDatasource(sqlFilePath, datasource);
+ submitMSQTask(sqlTask, datasource, msqContext);
+ }
+
+ /**
+ * Runs a SQL ingest test.
+ *
+ * @param sqlFilePath path of file containing the sql query.
+ * @param queryFilePath path of file containing the native test queries to be run on the ingested datasource.
+ * @param datasource name of the datasource. %%DATASOURCE%% in the sql and queries will be replaced with this value.
+ * @param msqContext context parameters to be passed with MSQ API call.
+ */
+ protected void runMSQTaskandTestQueries(String sqlFilePath,
+ String queryFilePath,
+ String datasource,
+ Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("Starting MSQ test for [%s, %s]", sqlFilePath, queryFilePath);
Review Comment:
Nit: the `[ ]` thing we use was probably originally for cases where the value is unformatted. Here, it would be cleaner to say:
```java
LOG.info("Starting MSQ test for sql path: %s, query path: %s", sqlFilePath, queryFilePath);
```
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -91,16 +97,149 @@ public String getFolderSuffix()
}
}
- private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
+ public static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Rule
+ public TestWatcher watchman = new TestWatcher()
+ {
+ @Override
+ public void starting(Description d)
+ {
+ LOG.info("RUNNING %s", d.getDisplayName());
+ }
+
+ @Override
+ public void failed(Throwable e, Description d)
+ {
+ LOG.error("FAILED %s", d.getDisplayName());
+ }
+
+ @Override
+ public void finished(Description d)
+ {
+ LOG.info("FINISHED %s", d.getDisplayName());
+ }
+ };
+
+ /**
+ * Reads file as utf-8 string and replace %%DATASOURCE%% with the provide datasource value.
+ */
+ protected String getStringFromFileAndReplaceDatasource(String filePath, String datasource)
+ {
+ String fileString;
+ try {
+ InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(filePath);
+ fileString = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ catch (IOException e) {
+ throw new ISE(e, "could not read query file: %s", filePath);
+ }
+
+ fileString = StringUtils.replace(
+ fileString,
+ "%%DATASOURCE%%",
+ datasource
+ );
+
+ return fileString;
+ }
+
+ /**
+ * Reads native queries from a file and runs against the provided datasource.
+ */
+ protected void doTestQuery(String dataSource, String queryFilePath)
+ {
+ try {
+ String query = getStringFromFileAndReplaceDatasource(queryFilePath, dataSource);
+ queryHelper.testQueriesFromString(query);
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while running test query at path " + queryFilePath);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTask(String sqlTask, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("SqlTask - \n %s", sqlTask);
+
+ // Submit the tasks and wait for the datasource to get loaded
+ msqHelper.submitMsqTaskAndWaitForCompletion(
+ sqlTask,
+ msqContext
+ );
+
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTaskFromFile(String sqlFilePath, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ String sqlTask = getStringFromFileAndReplaceDatasource(sqlFilePath, datasource);
+ submitMSQTask(sqlTask, datasource, msqContext);
+ }
+
+ /**
+ * Runs a SQL ingest test.
+ *
+ * @param sqlFilePath path of file containing the sql query.
+ * @param queryFilePath path of file containing the native test queries to be run on the ingested datasource.
+ * @param datasource name of the datasource. %%DATASOURCE%% in the sql and queries will be replaced with this value.
+ * @param msqContext context parameters to be passed with MSQ API call.
+ */
+ protected void runMSQTaskandTestQueries(String sqlFilePath,
+ String queryFilePath,
+ String datasource,
+ Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("Starting MSQ test for [%s, %s]", sqlFilePath, queryFilePath);
+
+ submitMSQTaskFromFile(sqlFilePath, datasource, msqContext);
+ doTestQuery(datasource, queryFilePath);
+ }
+
+ /**
+ * Runs a reindex SQL ingest test.
+ * Same as runMSQTaskandTestQueries, but replaces both %%DATASOURCE%% and %%REINDEX_DATASOURCE%% in the SQL Task.
+ */
+ protected void runReindexMSQTaskandTestQueries(String sqlFilePath,
+ String queryFilePath,
+ String datasource,
+ String reindexDatasource,
+ Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("Starting Reindex MSQ test for [%s, %s]", sqlFilePath, queryFilePath);
Review Comment:
See comments above for function signature, log line.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
Review Comment:
Does this take a list of `File` objects as well as a list of strings? If only a list of strings, then, sigh, Amazon is non-standard and the existing code is adequate, even if non-standard.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
+ }
+ catch (Exception e) {
+ // Fail if exception
+ fail();
+ }
+ }
+
+ @After
+ public void deleteSegmentsFromS3()
+ {
+ // Deleting folder created for storing segments (by druid) after test is completed
+ s3.deleteFolderFromS3(indexDatasource);
+ }
+
+ @AfterClass
+ public static void deleteDataFilesFromS3()
+ {
+ // Deleting uploaded data files
+ s3.deleteFilesFromS3(fileList());
+ }
+
+ void doTest(
+ Pair<String, List> s3InputSource,
+ Pair<Boolean, Boolean> segmentAvailabilityConfirmationPair
+ ) throws Exception
+ {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ try (
+ final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+ ) {
+ final Function<String, String> s3PropsTransform = spec -> {
+ try {
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%BUCKET%%",
+ config.getCloudBucket()
+ );
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%PATH%%",
+ config.getCloudPath()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_FORMAT_TYPE%%",
+ InputFormatDetails.JSON.getInputFormatType()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%PARTITIONS_SPEC%%",
+ jsonMapper.writeValueAsString(new DynamicPartitionsSpec(null, null))
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_TYPE%%",
+ "s3"
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ return StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+
+ doIndexTest(
+ indexDatasource,
+ INDEX_TASK,
+ s3PropsTransform,
+ INDEX_QUERIES_RESOURCE,
+ false,
+ true,
+ true,
+ segmentAvailabilityConfirmationPair
+ );
+ }
+ }
+
+ public void doMSQTest(Pair<String, List> s3InputSource,
+ String IngestSQLFilePath,
+ String TestQueriesFilePath)
+ {
+ try {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ String sqlTask = getStringFromFileAndReplaceDatasource(IngestSQLFilePath, indexDatasource);
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ Map<String, Object> context = ImmutableMap.of("finalizeAggregations", false,
+ "maxNumTasks", 5,
+ "groupByEnableMultiValueUnnesting", false);
+
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+
+ // Setting the correct object path in the sqlTask.
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%BUCKET%%",
+ config.getCloudBucket() // Getting from DRUID_CLOUD_BUCKET env variable
+ );
+ sqlTask = StringUtils.replace(
+ sqlTask,
+ "%%PATH%%",
+ config.getCloudPath() // Getting from DRUID_CLOUD_PATH env variable
+ );
+
+ submitMSQTask(sqlTask, indexDatasource, context);
+
+ // Verifying ingested datasource
+ doTestQuery(indexDatasource, TestQueriesFilePath);
+
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while testing [%s] with s3 input source property key [%s]",
+ IngestSQLFilePath, s3InputSource.lhs);
Review Comment:
Nit: in Java, names that start with a capital letter are classes. If this is a constant, then `INGEST_SQL_FILE_PATH`. If a variable, `ingestSqlFilePath`.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/ITS3ToS3ParallelIndexTest.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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 junitparams.Parameters;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+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;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 bucket name (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
Review Comment:
Great info, thanks!
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
Review Comment:
This is rather non-standard Java. For local files:
```java
List<File> filesToUpload = new ArrayList<>();
File localPath = new File("resources/data/batch_index/json/");
for (String file : fileList()) {
filesToUpload.add(new File(localPath, file));
}
```
If you want to be fancy:
```java
fileList().stream().map(new File(localPath, file))).collect(Collectors.toList());
```
I myself find the loop is easier to understand, but others on the team prefer the fancy approach.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
+ }
+ catch (Exception e) {
+ // Fail if exception
+ fail();
+ }
+ }
+
+ @After
+ public void deleteSegmentsFromS3()
+ {
+ // Deleting folder created for storing segments (by druid) after test is completed
+ s3.deleteFolderFromS3(indexDatasource);
+ }
+
+ @AfterClass
+ public static void deleteDataFilesFromS3()
+ {
+ // Deleting uploaded data files
+ s3.deleteFilesFromS3(fileList());
+ }
+
+ void doTest(
+ Pair<String, List> s3InputSource,
+ Pair<Boolean, Boolean> segmentAvailabilityConfirmationPair
+ ) throws Exception
+ {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ try (
+ final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+ ) {
+ final Function<String, String> s3PropsTransform = spec -> {
Review Comment:
Suggestion: pull this out as a real function so it can be more easily tested. It is awkward to do the rewrites only as part of job submission.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
+ }
+ catch (Exception e) {
+ // Fail if exception
+ fail();
+ }
+ }
+
+ @After
+ public void deleteSegmentsFromS3()
+ {
+ // Deleting folder created for storing segments (by druid) after test is completed
+ s3.deleteFolderFromS3(indexDatasource);
+ }
+
+ @AfterClass
+ public static void deleteDataFilesFromS3()
+ {
+ // Deleting uploaded data files
+ s3.deleteFilesFromS3(fileList());
+ }
+
+ void doTest(
+ Pair<String, List> s3InputSource,
+ Pair<Boolean, Boolean> segmentAvailabilityConfirmationPair
+ ) throws Exception
+ {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ try (
+ final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+ ) {
+ final Function<String, String> s3PropsTransform = spec -> {
+ try {
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%BUCKET%%",
+ config.getCloudBucket()
+ );
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%PATH%%",
+ config.getCloudPath()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_FORMAT_TYPE%%",
+ InputFormatDetails.JSON.getInputFormatType()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%PARTITIONS_SPEC%%",
+ jsonMapper.writeValueAsString(new DynamicPartitionsSpec(null, null))
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_TYPE%%",
+ "s3"
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ return StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+
+ doIndexTest(
+ indexDatasource,
+ INDEX_TASK,
+ s3PropsTransform,
+ INDEX_QUERIES_RESOURCE,
+ false,
+ true,
+ true,
+ segmentAvailabilityConfirmationPair
+ );
+ }
+ }
+
+ public void doMSQTest(Pair<String, List> s3InputSource,
Review Comment:
Here and elsewhere. Would be great to include a Javadoc comment explaining what this does. Yes, we can figure it out. But, since you wrote it, you could save us time by explaining what your intent is for this function. That is, does it just run? Does it run and check the results? Where are the input files? What are the parameters we should put into those files? Where do the values come from? Some of this can be at the to of the test. If it is shared across a suite of tests, please put it in the base class.
Explaining here is best. There should also be some basic info in the doc files, which can point here for details.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
+ }
+ catch (Exception e) {
+ // Fail if exception
+ fail();
+ }
+ }
+
+ @After
+ public void deleteSegmentsFromS3()
+ {
+ // Deleting folder created for storing segments (by druid) after test is completed
+ s3.deleteFolderFromS3(indexDatasource);
+ }
+
+ @AfterClass
+ public static void deleteDataFilesFromS3()
+ {
+ // Deleting uploaded data files
+ s3.deleteFilesFromS3(fileList());
+ }
+
+ void doTest(
+ Pair<String, List> s3InputSource,
+ Pair<Boolean, Boolean> segmentAvailabilityConfirmationPair
+ ) throws Exception
+ {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ try (
+ final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+ ) {
+ final Function<String, String> s3PropsTransform = spec -> {
+ try {
+ String inputSourceValue = jsonMapper.writeValueAsString(s3InputSource.rhs);
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%BUCKET%%",
+ config.getCloudBucket()
+ );
+ inputSourceValue = StringUtils.replace(
+ inputSourceValue,
+ "%%PATH%%",
+ config.getCloudPath()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_FORMAT_TYPE%%",
+ InputFormatDetails.JSON.getInputFormatType()
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%PARTITIONS_SPEC%%",
+ jsonMapper.writeValueAsString(new DynamicPartitionsSpec(null, null))
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_TYPE%%",
+ "s3"
+ );
+ spec = StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_KEY%%",
+ s3InputSource.lhs
+ );
+ return StringUtils.replace(
+ spec,
+ "%%INPUT_SOURCE_PROPERTY_VALUE%%",
+ inputSourceValue
+ );
+ }
+ catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+
+ doIndexTest(
+ indexDatasource,
+ INDEX_TASK,
+ s3PropsTransform,
+ INDEX_QUERIES_RESOURCE,
+ false,
+ true,
+ true,
+ segmentAvailabilityConfirmationPair
+ );
+ }
+ }
+
+ public void doMSQTest(Pair<String, List> s3InputSource,
+ String IngestSQLFilePath,
+ String TestQueriesFilePath)
Review Comment:
Nit: method signature format.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
Review Comment:
A Javadoc comment would be helpful to explain what this does. The original didn't have one, but now that you've figured it out, would be good to explain the purpose to the rest of us.
##########
integration-tests/src/main/java/org/apache/druid/testing/utils/S3TestUtil.java:
##########
@@ -0,0 +1,171 @@
+/*
+ * 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.testing.utils;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest;
+import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
+import com.amazonaws.services.s3.model.ListObjectsRequest;
+import com.amazonaws.services.s3.model.ObjectListing;
+import com.amazonaws.services.s3.model.S3ObjectSummary;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+public class S3TestUtil
Review Comment:
I agree. @cryptoe, you created such an API for Druid's internal use. Can we reuse that here? Create a test-specific wrapper? The idea state would be that there is an `ObjectStoreClient` interface which provides the upload, delete, etc. methods. @cryptoe, can you help @abhagraw define such a test client? Doing so goes a bit beyond the "convert the old IT" task and would benefit from some dev help.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractS3InputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 Bucket to store in (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in
+ * AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
Review Comment:
This comment is duplicated, which is perhaps OK. The commonality is the `S3DeepStorage` test category. Perhaps it makes sense to move the documentation there for all tests that run with that configuration?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractS3InputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 Bucket to store in (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in
+ * AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractS3InputSourceParallelIndexTest
Review Comment:
As noted earlier, a brief comment would be helpful to us reviewers. I gather this is the MSQ ingestion test, but using S3 as deep storage? Also as the source of the files?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractS3InputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 Bucket to store in (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in
+ * AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
+ */
+
+@RunWith(DruidTestRunner.class)
+@Category(S3DeepStorage.class)
+public class ITS3SQLBasedIngestionTest extends AbstractS3InputSourceParallelIndexTest
+{
+ @Inject
+ @Json
+ protected ObjectMapper jsonMapper;
+ private static final String CLOUD_INGEST_SQL = "/multi-stage-query/wikipedia_cloud_index_msq.sql";
+ private static final String INDEX_QUERIES_FILE = "/multi-stage-query/wikipedia_index_queries.json";
Review Comment:
The typical Java standard is to put constants at the top of the file, followed by variables. So, move `jsonMapper` after these two constants, with a blank line between them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1052936091
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
Review Comment:
If allowed, I would like to keep the for loop - since it is much for readable for Java newbies like me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on PR #13535:
URL: https://github.com/apache/druid/pull/13535#issuecomment-1354263606
@paul-rogers Have updated the code as per comments.
I have moved most of the s3 related code to a separate s3TestUtil which is being used by both `ITS3SQLBasedIngestionTest` and `AbstractS3InputSourceParallelIndexTest`.
There will be some amount of code duplication because msq related tests need to inherit from `AbstractITSQLBasedIngestionTest` while the test being migrated from old IT framework have a hierarchical structure already in place.
Also, I would recommend you to checkout `AbstractITSQLBasedIngestion ` class - which contains most of the methods we use in this test with doc strings. Not sure - if we want to have the same being mentioned in each test. (We are verifying the results after ingesting data).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: MSQ s3 deep storage tests
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1052938244
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
+ }
+ catch (Exception e) {
+ // Fail if exception
+ fail();
+ }
+ }
+
+ @After
+ public void deleteSegmentsFromS3()
+ {
+ // Deleting folder created for storing segments (by druid) after test is completed
+ s3.deleteFolderFromS3(indexDatasource);
+ }
+
+ @AfterClass
+ public static void deleteDataFilesFromS3()
+ {
+ // Deleting uploaded data files
+ s3.deleteFilesFromS3(fileList());
+ }
+
+ void doTest(
+ Pair<String, List> s3InputSource,
+ Pair<Boolean, Boolean> segmentAvailabilityConfirmationPair
+ ) throws Exception
+ {
+ indexDatasource = "wikipedia_index_test_" + UUID.randomUUID();
+ try (
+ final Closeable ignored1 = unloader(indexDatasource + config.getExtraDatasourceNameSuffix());
+ ) {
+ final Function<String, String> s3PropsTransform = spec -> {
Review Comment:
I though of the same as this makes debugging difficult.
But we are passing the function as arguments across multiple doTest methods (which are being used by a lot of other tests). So deciding against changing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] abhagraw commented on a diff in pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
abhagraw commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1053008004
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/msq/ITS3SQLBasedIngestionTest.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * 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.msq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.inject.Inject;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.apache.druid.guice.annotations.Json;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.testsEx.categories.S3DeepStorage;
+import org.apache.druid.testsEx.config.DruidTestRunner;
+import org.apache.druid.testsEx.indexer.AbstractS3InputSourceParallelIndexTest;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.util.List;
+
+/**
+ * IMPORTANT:
+ * To run this test, you must set the following env variables in the build environment
+ * DRUID_CLOUD_BUCKET - s3 Bucket to store in (value to be set in druid.storage.bucket)
+ * DRUID_CLOUD_PATH - path inside the bucket where the test data files will be uploaded
+ * (this will also be used as druid.storage.baseKey for s3 deep storage setup)
+ * <p>
+ * The AWS key, secret and region should be set in
+ * AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and AWS_REGION respectively.
+ * <p>
+ * <a href="https://druid.apache.org/docs/latest/development/extensions-core/s3.html">S3 Deep Storage setup in druid</a>
Review Comment:
I feel having it near the test is better. Don't see any reason for people to look at the S3DeepStorage test category file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
[GitHub] [druid] paul-rogers commented on a diff in pull request #13535: Cloud deep storage tests in new IT framework (S3, GCS, Azure)
Posted by GitBox <gi...@apache.org>.
paul-rogers commented on code in PR #13535:
URL: https://github.com/apache/druid/pull/13535#discussion_r1053774387
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/config/IntegrationTestingConfigEx.java:
##########
@@ -353,6 +354,18 @@ public String getCloudBucket()
return getProperty("cloudBucket");
}
+ @Override
+ public String getCloudBucketwhenDeepStorageTypeIs(String storageType)
+ {
+ if (Objects.equals(storageType, "s3")) {
+ return getProperty("cloudBucket");
+ } else if (Objects.equals(storageType, "azure")) {
Review Comment:
Might want to declare constants for these strings so that we can be sure that the caller and this method use the same strings.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ public String setInputSourceInPath(String inputSourceType, String inputSourcePath)
+ {
+ if ("google".equals(inputSourceType)) {
Review Comment:
Please pull "google" out as a constant.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
+ }
+ try {
+ s3 = new S3TestUtil();
+ s3.uploadDataFilesToS3(filesToUpload);
Review Comment:
I suppose this code won't run anywhere except MacOS/Linux, so it is probably OK to hard-code assumptions about paths...
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
Review Comment:
Nit: this can be a constant: it doesn't vary per instance.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/config/IntegrationTestingConfigEx.java:
##########
@@ -353,6 +354,18 @@ public String getCloudBucket()
return getProperty("cloudBucket");
}
+ @Override
+ public String getCloudBucketwhenDeepStorageTypeIs(String storageType)
+ {
+ if (Objects.equals(storageType, "s3")) {
Review Comment:
`"s3".equals(storageType)` is the typical Java pattern.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractITBatchIndexTest.java:
##########
@@ -93,14 +99,149 @@ public String getFolderSuffix()
private static final Logger LOG = new Logger(AbstractITBatchIndexTest.class);
- @Inject
- protected IntegrationTestingConfig config;
@Inject
protected SqlTestQueryHelper sqlQueryHelper;
@Inject
ClientInfoResourceTestClient clientInfoResourceTestClient;
+ @Inject
+ private MsqTestQueryHelper msqHelper;
+
+ @Inject
+ protected TestQueryHelper queryHelper;
+
+ @Inject
+ private DataLoaderHelper dataLoaderHelper;
+
+ @Rule
+ public TestWatcher watchman = new TestWatcher()
+ {
+ @Override
+ public void starting(Description d)
+ {
+ LOG.info("RUNNING %s", d.getDisplayName());
+ }
+
+ @Override
+ public void failed(Throwable e, Description d)
+ {
+ LOG.error("FAILED %s", d.getDisplayName());
+ }
+
+ @Override
+ public void finished(Description d)
+ {
+ LOG.info("FINISHED %s", d.getDisplayName());
+ }
+ };
+
+ /**
+ * Reads file as utf-8 string and replace %%DATASOURCE%% with the provide datasource value.
+ */
+ public String getStringFromFileAndReplaceDatasource(String filePath, String datasource)
+ {
+ String fileString;
+ try {
+ InputStream is = AbstractITBatchIndexTest.class.getResourceAsStream(filePath);
+ fileString = IOUtils.toString(is, StandardCharsets.UTF_8);
+ }
+ catch (IOException e) {
+ throw new ISE(e, "could not read query file: %s", filePath);
+ }
+
+ fileString = StringUtils.replace(
+ fileString,
+ "%%DATASOURCE%%",
+ datasource
+ );
+
+ return fileString;
+ }
+
+ /**
+ * Reads native queries from a file and runs against the provided datasource.
+ */
+ protected void doTestQuery(String dataSource, String queryFilePath)
+ {
+ try {
+ String query = getStringFromFileAndReplaceDatasource(queryFilePath, dataSource);
+ queryHelper.testQueriesFromString(query);
+ }
+ catch (Exception e) {
+ LOG.error(e, "Error while running test query at path " + queryFilePath);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Submits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTask(String sqlTask, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ LOG.info("SqlTask - \n %s", sqlTask);
+
+ // Submit the tasks and wait for the datasource to get loaded
+ msqHelper.submitMsqTaskAndWaitForCompletion(
+ sqlTask,
+ msqContext
+ );
+
+ dataLoaderHelper.waitUntilDatasourceIsReady(datasource);
+ }
+
+ /**
+ * Sumits a sqlTask, waits for task completion.
+ */
+ protected void submitMSQTaskFromFile(String sqlFilePath, String datasource, Map<String, Object> msqContext) throws Exception
+ {
+ String sqlTask = getStringFromFileAndReplaceDatasource(sqlFilePath, datasource);
+ submitMSQTask(sqlTask, datasource, msqContext);
+ }
+
+ /**
+ * Runs a SQL ingest test.
+ *
+ * @param sqlFilePath path of file containing the sql query.
+ * @param queryFilePath path of file containing the native test queries to be run on the ingested datasource.
+ * @param datasource name of the datasource. %%DATASOURCE%% in the sql and queries will be replaced with this value.
+ * @param msqContext context parameters to be passed with MSQ API call.
+ */
+ protected void runMSQTaskandTestQueries(String sqlFilePath,
+ String queryFilePath,
+ String datasource,
+ Map<String, Object> msqContext
+ ) throws Exception
+ {
+ LOG.info("Starting MSQ test for sql path:%s, query path: %s]", sqlFilePath, queryFilePath);
Review Comment:
Nit: space after first `:`: `path: %s`. Here and below.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/config/IntegrationTestingConfigEx.java:
##########
@@ -353,6 +354,18 @@ public String getCloudBucket()
return getProperty("cloudBucket");
}
+ @Override
+ public String getCloudBucketwhenDeepStorageTypeIs(String storageType)
+ {
+ if (Objects.equals(storageType, "s3")) {
+ return getProperty("cloudBucket");
+ } else if (Objects.equals(storageType, "azure")) {
+ return getProperty("azureContainer");
+ } else {
Review Comment:
Should probably explicitly check for Google, then decide what to do if the value is none of the above. Return null? Or, throw an error?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "%%INPUT_SOURCE_TYPE%%://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ public String setInputSourceInPath(String inputSourceType, String inputSourcePath)
+ {
+ if ("google".equals(inputSourceType)) {
+ return StringUtils.replace(
+ inputSourcePath,
+ "%%INPUT_SOURCE_TYPE%%",
+ "gs"
+ );
+ } else {
+ return StringUtils.replace(
+ inputSourcePath,
+ "%%INPUT_SOURCE_TYPE%%",
+ inputSourceType
+ );
+ }
+ }
+
+ public String getCloudPath(String inputSourceType)
+ {
+ if ("google".equals(inputSourceType)) {
+ return config.getProperty("googlePrefix");
Review Comment:
Please pull property names out as constants.
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractCloudInputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+/**
+ * Common abstract class for most cloud deep storage tests
+ */
+public abstract class AbstractCloudInputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+ private static final Logger LOG = new Logger(AbstractCloudInputSourceParallelIndexTest.class);
+
+ String indexDatasource = "wikipedia_cloud_index_test_";
+
+ public static Object[][] resources()
Review Comment:
Isn't the return value `Pair<String, List<?>>[][]`?
##########
integration-tests-ex/cases/src/test/java/org/apache/druid/testsEx/indexer/AbstractS3InputSourceParallelIndexTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.testing.utils.S3TestUtil;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.io.Closeable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Function;
+
+import static junit.framework.Assert.fail;
+
+
+public abstract class AbstractS3InputSourceParallelIndexTest extends AbstractITBatchIndexTest
+{
+ private static String indexDatasource;
+ private static S3TestUtil s3;
+ private static final String INDEX_TASK = "/indexer/wikipedia_cloud_index_task.json";
+ private static final String INDEX_QUERIES_RESOURCE = "/indexer/wikipedia_index_queries.json";
+ private static final String INPUT_SOURCE_URIS_KEY = "uris";
+ private static final String INPUT_SOURCE_PREFIXES_KEY = "prefixes";
+ private static final String INPUT_SOURCE_OBJECTS_KEY = "objects";
+ private static final String WIKIPEDIA_DATA_1 = "wikipedia_index_data1.json";
+ private static final String WIKIPEDIA_DATA_2 = "wikipedia_index_data2.json";
+ private static final String WIKIPEDIA_DATA_3 = "wikipedia_index_data3.json";
+
+ public static Object[][] resources()
+ {
+ return new Object[][]{
+ {new Pair<>(INPUT_SOURCE_URIS_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_1,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_2,
+ "s3://%%BUCKET%%/%%PATH%%/" + WIKIPEDIA_DATA_3
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_PREFIXES_KEY,
+ ImmutableList.of(
+ "s3://%%BUCKET%%/%%PATH%%/"
+ )
+ )},
+ {new Pair<>(INPUT_SOURCE_OBJECTS_KEY,
+ ImmutableList.of(
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_1),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_2),
+ ImmutableMap.of("bucket", "%%BUCKET%%", "path", "%%PATH%%/" + WIKIPEDIA_DATA_3)
+ )
+ )}
+ };
+ }
+
+ public static List<String> fileList()
+ {
+ return Arrays.asList(WIKIPEDIA_DATA_1, WIKIPEDIA_DATA_2, WIKIPEDIA_DATA_3);
+ }
+
+ @BeforeClass
+ public static void uploadDataFilesToS3()
+ {
+ List<String> filesToUpload = new ArrayList<>();
+ String localPath = "resources/data/batch_index/json/";
+ for (String file : fileList()) {
+ filesToUpload.add(localPath + file);
Review Comment:
It is more readable even for old-timers like me. However, I always get feedback asking me to change loops to use streaming, which I find harder to read. Since I'm the only one review this, let's just keep it a loop unless someone else complains.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org