PHOENIX-6430 Added support for full row update for tables when no columns specfied in scenario

phoenix-pherf/src/it/java/org/apache/phoenix/pherf/workload/mt/tenantoperation/MultiTenantTableOperationBaseIT.java

@@ -0,0 +1,74 @@
+/*
+ * 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.phoenix.pherf.workload.mt.tenantoperation;
+
+import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
+import org.apache.phoenix.pherf.PherfConstants;
+import org.apache.phoenix.pherf.XMLConfigParserTest;
+import org.apache.phoenix.pherf.configuration.DataModel;
+import org.apache.phoenix.pherf.configuration.XMLConfigParser;
+import org.apache.phoenix.pherf.schema.SchemaReader;
+import org.apache.phoenix.pherf.util.PhoenixUtil;
+import org.junit.BeforeClass;
+
+import java.net.URL;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+public class MultiTenantTableOperationBaseIT extends ParallelStatsDisabledIT {
+
+ protected static final String matcherScenario = ".*scenario/.*test_tbl_workload.*xml";
+ protected static final String matcherSchema = ".*datamodel/.*test_schema_tbl*.*sql";
+
+ protected static PhoenixUtil util = PhoenixUtil.create(true);
+ protected static Properties properties;
+ protected static SchemaReader reader;
+ protected static XMLConfigParser parser;
+ protected static List<Path> resources;
+
+ @BeforeClass public static synchronized void setUp() throws Exception {
+ PherfConstants constants = PherfConstants.create();
+ properties = constants.getProperties(PherfConstants.PHERF_PROPERTIES, false);
+
+ PhoenixUtil.setZookeeper("localhost");
+ reader = new SchemaReader(util, matcherSchema);
+ parser = new XMLConfigParser(matcherScenario);
+ reader.applySchema();
+ resources = new ArrayList<>(reader.getResourceList());
+
+ assertTrue("Could not pull list of schema files.", resources.size() > 0);
+ assertNotNull("Could not read schema file.", reader.resourceToString(resources.get(0)));
+
+ }
+
+ public DataModel readTestDataModel(String resourceName) throws Exception {
+ URL scenarioUrl = XMLConfigParserTest.class.getResource(resourceName);
+ assertNotNull(scenarioUrl);
+ Path p = Paths.get(scenarioUrl.toURI());
+ return XMLConfigParser.readDataModel(p);
+ }
+
+}

phoenix-pherf/src/it/java/org/apache/phoenix/pherf/workload/mt/tenantoperation/MultiTenantTestUtils.java

@@ -0,0 +1,267 @@
+/*
+ * 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.phoenix.pherf.workload.mt.tenantoperation;
+
+import com.lmax.disruptor.LifecycleAware;
+import com.lmax.disruptor.WorkHandler;
+import org.apache.phoenix.pherf.configuration.DataModel;
+import org.apache.phoenix.pherf.configuration.LoadProfile;
+import org.apache.phoenix.pherf.configuration.Scenario;
+import org.apache.phoenix.pherf.util.PhoenixUtil;
+import org.apache.phoenix.pherf.workload.Workload;
+import org.apache.phoenix.pherf.workload.mt.Operation;
+import org.apache.phoenix.pherf.workload.mt.OperationStats;
+import org.apache.phoenix.thirdparty.com.google.common.base.Function;
+import org.apache.phoenix.thirdparty.com.google.common.base.Supplier;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Lists;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+import org.junit.Assert;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class MultiTenantTestUtils {
+ private static final Logger LOGGER = LoggerFactory.getLogger(MultiTenantTestUtils.class);
+ enum TestOperationGroup {
+ upsertOp, queryOp1, queryOp2, idleOp, udfOp
+ }
+
+ public void testWorkloadWithOneHandler(Properties properties, DataModel model,
+ String scenarioName, int expectedTenantGroups, int expectedOpGroups) throws Exception {
+
+ int numHandlers = 1;
+ int totalOperations = 50;
+ int perHandlerCount = 50;
+
+ ExecutorService executor = null;
+ try {
+ executor = Executors.newFixedThreadPool(numHandlers);
+ PhoenixUtil pUtil = PhoenixUtil.create();
+ for (Scenario scenario : model.getScenarios()) {
+ if (scenarioName != null && !scenarioName.isEmpty()
+ && scenario.getName().compareTo(scenarioName) != 0) {
+ continue;
+ }
+ LOGGER.debug(String.format("Testing %s", scenario.getName()));
+ LoadProfile loadProfile = scenario.getLoadProfile();
+
+ // Set the total number of operations for this load profile
+ loadProfile.setNumOperations(totalOperations);
+ TenantOperationFactory opFactory = new TenantOperationFactory(pUtil, model,
+ scenario);
+ assertEquals("tenant group size is not as expected: ", expectedTenantGroups,
+ loadProfile.getTenantDistribution().size());
+ assertEquals("operation group size from the factory is not as expected: ",
+ expectedOpGroups, opFactory.getOperations().size());
+
+ // populate the handlers and countdown latches.
+ String handlerId = String
+ .format("%s.%d", InetAddress.getLocalHost().getHostName(), numHandlers);
+ List<WorkHandler> workers = Lists.newArrayList();
+ Map<String, CountDownLatch> latches = Maps.newConcurrentMap();
+ workers.add(new EventCountingWorkHandler(opFactory, handlerId, latches));
+ latches.put(handlerId, new CountDownLatch(perHandlerCount));
+ // submit the workload
+ Workload workload = new TenantOperationWorkload(pUtil, model, scenario, workers,
+ properties);
+ Future status = executor.submit(workload.execute());
+ // Just make sure there are no exceptions
+ status.get();
+
+ // Wait for the handlers to count down
+ for (Map.Entry<String, CountDownLatch> latch : latches.entrySet()) {
+ assertTrue(latch.getValue().await(60, TimeUnit.SECONDS));
+ }
+ }
+ } finally {
+ if (executor != null) {
+ executor.shutdown();
+ }
+ }
+ }
+
+ public void testWorkloadWithManyHandlers(Properties properties, DataModel model,
+ String scenarioName, int expectedTenantGroups, int expectedOpGroups) throws Exception {
+
+ int numHandlers = 5;
+ int totalOperations = 500;
+ int perHandlerCount = 50;
+
+ ExecutorService executor = Executors.newFixedThreadPool(numHandlers);
+ PhoenixUtil pUtil = PhoenixUtil.create();
+ for (Scenario scenario : model.getScenarios()) {
+ if (scenarioName != null && !scenarioName.isEmpty()
+ && scenario.getName().compareTo(scenarioName) != 0) {
+ continue;
+ }
+ LOGGER.debug(String.format("Testing %s", scenario.getName()));
+ LoadProfile loadProfile = scenario.getLoadProfile();
+
+ // Set the total number of operations for this load profile
+ loadProfile.setNumOperations(totalOperations);
+ TenantOperationFactory opFactory = new TenantOperationFactory(pUtil, model, scenario);
+ assertEquals("tenant group size is not as expected: ", expectedTenantGroups,
+ loadProfile.getTenantDistribution().size());
+
+ assertEquals("operation group size from the factory is not as expected: ",
+ expectedOpGroups, opFactory.getOperations().size());
+
+ // populate the handlers and countdown latches.
+ List<WorkHandler> workers = Lists.newArrayList();
+ Map<String, CountDownLatch> latches = Maps.newConcurrentMap();
+ for (int i = 0; i < numHandlers; i++) {
+ String handlerId = String
+ .format("%s.%d", InetAddress.getLocalHost().getHostName(), i);
+ workers.add(new EventCountingWorkHandler(opFactory, handlerId, latches));
+ latches.put(handlerId, new CountDownLatch(perHandlerCount));
+ }
+ // submit the workload
+ Workload workload = new TenantOperationWorkload(pUtil, model, scenario, workers,
+ properties);
+ Future status = executor.submit(workload.execute());
+ // Just make sure there are no exceptions
+ status.get();
+ // Wait for the handlers to count down
+ for (Map.Entry<String, CountDownLatch> latch : latches.entrySet()) {
+ assertTrue(latch.getValue().await(60, TimeUnit.SECONDS));
+ }
+ }
+ executor.shutdown();
+ }
+
+ public void testVariousOperations(Properties properties, DataModel model, String scenarioName,
+ int expectedTenantGroups, int expectedOpGroups) throws Exception {
+
+ int numRuns = 10;
+ int numOperations = 10;
+
+ PhoenixUtil pUtil = PhoenixUtil.create();
+ for (Scenario scenario : model.getScenarios()) {
+ if (scenarioName != null && !scenarioName.isEmpty()
+ && scenario.getName().compareTo(scenarioName) != 0) {
+ continue;
+ }
+ LOGGER.debug(String.format("Testing %s", scenario.getName()));
+ LoadProfile loadProfile = scenario.getLoadProfile();
+ assertEquals("tenant group size is not as expected: ", expectedTenantGroups,
+ loadProfile.getTenantDistribution().size());
+ assertEquals("operation group size is not as expected: ", expectedOpGroups,
+ loadProfile.getOpDistribution().size());
+
+ TenantOperationFactory opFactory = new TenantOperationFactory(pUtil, model, scenario);
+ TenantOperationEventGenerator evtGen = new TenantOperationEventGenerator(
+ opFactory.getOperations(), model, scenario);
+
+ assertEquals("operation group size from the factory is not as expected: ",
+ expectedOpGroups, opFactory.getOperations().size());
+
+ int numRowsInserted = 0;
+ for (int i = 0; i < numRuns; i++) {
+ int ops = numOperations;
+ loadProfile.setNumOperations(ops);
+ while (ops-- > 0) {
+ TenantOperationInfo info = evtGen.next();
+ Supplier<Function<TenantOperationInfo, OperationStats>> opSupplier = opFactory
+ .getOperationSupplier(info);
+ OperationStats stats = opSupplier.get().apply(info);
+ LOGGER.info(PhoenixUtil.getGSON().toJson(stats));
+ if (info.getOperation().getType() == Operation.OperationType.PRE_RUN) continue;
+ assertTrue(stats.getStatus() != -1);
+ switch (TestOperationGroup
+ .valueOf(info.getOperationGroupId())) {
+ case upsertOp:
+ assertTrue(opSupplier.getClass()
+ .isAssignableFrom(UpsertOperationSupplier.class));
+ numRowsInserted += stats.getRowCount();
+ break;
+ case queryOp1:
+ case queryOp2:
+ assertTrue(opFactory.getOperationSupplier(info).getClass()
+ .isAssignableFrom(QueryOperationSupplier.class));
+
+ // expected row count > 0
+ // Since the same view/table is being used by many tests.
+ // Keeping query return values would require lot of housekeeping
+ assertTrue(stats.getRowCount() > 0);
+ break;
+ case idleOp:
+ assertTrue(opFactory.getOperationSupplier(info).getClass()
+ .isAssignableFrom(IdleTimeOperationSupplier.class));
+ assertEquals(0, stats.getRowCount());
+ // expected think time (no-op) to be ~50ms
+ assertTrue(40 < stats.getDurationInMs() && stats.getDurationInMs() < 60);
+ break;
+ case udfOp:
+ assertTrue(opFactory.getOperationSupplier(info).getClass()
+ .isAssignableFrom(UserDefinedOperationSupplier.class));
+ assertEquals(0, stats.getRowCount());
+ break;
+ default:
+ Assert.fail();
+ }
+ }
+ }
+ }
+ }
+
+ private static class EventCountingWorkHandler
+ implements WorkHandler<TenantOperationWorkload.TenantOperationEvent>, LifecycleAware {
+ private final String handlerId;
+ private final TenantOperationFactory tenantOperationFactory;
+ private final Map<String, CountDownLatch> latches;
+
+ public EventCountingWorkHandler(TenantOperationFactory tenantOperationFactory,
+ String handlerId, Map<String, CountDownLatch> latches) {
+ this.handlerId = handlerId;
+ this.tenantOperationFactory = tenantOperationFactory;
+ this.latches = latches;
+ }
+
+ @Override public void onStart() {
+ }
+
+ @Override public void onShutdown() {
+ }
+
+ @Override public void onEvent(TenantOperationWorkload.TenantOperationEvent event)
+ throws Exception {
+ TenantOperationInfo input = event.getTenantOperationInfo();
+ Supplier<Function<TenantOperationInfo, OperationStats>>
+ opSupplier
+ = tenantOperationFactory.getOperationSupplier(input);
+ OperationStats stats = opSupplier.get().apply(input);
+ LOGGER.info(PhoenixUtil.getGSON().toJson(stats));
+ assertEquals(0, stats.getStatus());
+ latches.get(handlerId).countDown();
+ }
+ }
+
+}

phoenix-pherf/src/it/java/org/apache/phoenix/pherf/workload/mt/tenantoperation/MultiTenantOperationBaseIT.java → phoenix-pherf/src/it/java/org/apache/phoenix/pherf/workload/mt/tenantoperation/MultiTenantViewOperationBaseIT.java

@@ -19,7 +19,6 @@
 
 package org.apache.phoenix.pherf.workload.mt.tenantoperation;
 
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
 import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
 import org.apache.phoenix.pherf.PherfConstants;
@@ -28,7 +27,6 @@
 import org.apache.phoenix.pherf.configuration.XMLConfigParser;
 import org.apache.phoenix.pherf.schema.SchemaReader;
 import org.apache.phoenix.pherf.util.PhoenixUtil;
-import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.experimental.categories.Category;
@@ -44,14 +42,7 @@
 import static org.junit.Assert.assertTrue;
 
 @Category(NeedsOwnMiniClusterTest.class)
-public class MultiTenantOperationBaseIT extends ParallelStatsDisabledIT {
- enum TestOperationGroup {
- upsertOp, queryOp1, queryOp2, idleOp, udfOp
- }
-
- static enum TestTenantGroup {
- tg1, tg2, tg3
- }
+public class MultiTenantViewOperationBaseIT extends ParallelStatsDisabledIT {
  protected static final String matcherScenario = ".*scenario/.*test_mt_workload.*xml";
  protected static final String matcherSchema = ".*datamodel/.*test_schema_mt*.*sql";