You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ka...@apache.org on 2019/01/30 22:16:39 UTC
[phoenix] branch master updated: PHOENIX-4009 Run UPDATE STATISTICS
command by using MR integration on snapshots
This is an automated email from the ASF dual-hosted git repository.
karanmehta93 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new bcbcfd8 PHOENIX-4009 Run UPDATE STATISTICS command by using MR integration on snapshots
bcbcfd8 is described below
commit bcbcfd817436921de6c69af4cb9a10ce1adc51ac
Author: Karan Mehta <ka...@gmail.com>
AuthorDate: Wed Jan 30 14:08:54 2019 -0800
PHOENIX-4009 Run UPDATE STATISTICS command by using MR integration on snapshots
---
phoenix-core/pom.xml | 8 +
...olumnEncodedImmutableNonTxStatsCollectorIT.java | 39 ---
.../ColumnEncodedImmutableTxStatsCollectorIT.java | 42 ---
.../ColumnEncodedMutableTxStatsCollectorIT.java | 41 ---
...onColumnEncodedImmutableTxStatsCollectorIT.java | 42 ---
.../phoenix/end2end/ParallelStatsDisabledIT.java | 16 +-
.../apache/phoenix/end2end/SpillableGroupByIT.java | 2 +-
...sCollectorIT.java => BaseStatsCollectorIT.java} | 243 +++++++-------
.../stats/NamespaceDisabledStatsCollectorIT.java} | 28 +-
.../stats/NamespaceEnabledStatsCollectorIT.java} | 47 ++-
.../phoenix/schema/stats/NoOpStatsCollectorIT.java | 143 +++++++++
.../stats/NonTxStatsCollectorIT.java} | 26 +-
.../phoenix/schema/stats/TxStatsCollectorIT.java | 51 +++
.../UngroupedAggregateRegionObserver.java | 67 ++--
.../apache/phoenix/exception/SQLExceptionCode.java | 2 +
.../apache/phoenix/iterate/SnapshotScanner.java | 54 +++-
.../iterate/TableSnapshotResultIterator.java | 21 +-
.../phoenix/mapreduce/PhoenixInputFormat.java | 50 +--
.../phoenix/mapreduce/PhoenixRecordReader.java | 5 +-
.../mapreduce/util/PhoenixConfigurationUtil.java | 40 ++-
.../mapreduce/util/PhoenixMapReduceUtil.java | 13 +-
.../org/apache/phoenix/query/GuidePostsCache.java | 4 +-
.../org/apache/phoenix/query/QueryServices.java | 3 -
.../org/apache/phoenix/schema/MetaDataClient.java | 16 +-
.../schema/stats/DefaultStatisticsCollector.java | 355 +++++++++++----------
.../schema/stats/NoOpStatisticsCollector.java | 24 +-
.../phoenix/schema/stats/StatisticsCollector.java | 25 +-
.../schema/stats/StatisticsCollectorFactory.java | 19 +-
.../phoenix/schema/stats/StatisticsScanner.java | 3 +-
.../phoenix/schema/stats/StatisticsUtil.java | 23 +-
.../phoenix/schema/stats/StatisticsWriter.java | 60 +++-
.../StatsCollectionDisabledOnServerException.java | 33 ++
.../phoenix/schema/stats/UpdateStatisticsTool.java | 223 +++++++++++++
.../java/org/apache/phoenix/util/ServerUtil.java | 3 +-
.../util/PhoenixConfigurationUtilTest.java | 17 +
.../phoenix/pig/util/PhoenixPigSchemaUtil.java | 4 +-
pom.xml | 10 +
37 files changed, 1165 insertions(+), 637 deletions(-)
diff --git a/phoenix-core/pom.xml b/phoenix-core/pom.xml
index db5b909..2496ce1 100644
--- a/phoenix-core/pom.xml
+++ b/phoenix-core/pom.xml
@@ -418,6 +418,14 @@
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-metrics-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-metrics</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
<scope>test</scope>
<type>test-jar</type>
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableNonTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableNonTxStatsCollectorIT.java
deleted file mode 100644
index 0481ba5..0000000
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableNonTxStatsCollectorIT.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.end2end;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
-import org.junit.runners.Parameterized.Parameters;
-
-public class ColumnEncodedImmutableNonTxStatsCollectorIT extends StatsCollectorIT {
-
- public ColumnEncodedImmutableNonTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
- }
-
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
- return Arrays.asList(new Object[][] {
- { false, null, false, true }, { false, null, true, true }
- });
- }
-}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableTxStatsCollectorIT.java
deleted file mode 100644
index 073dc07..0000000
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedImmutableTxStatsCollectorIT.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.end2end;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
-import org.apache.phoenix.util.TestUtil;
-import org.junit.runners.Parameterized.Parameters;
-
-public class ColumnEncodedImmutableTxStatsCollectorIT extends StatsCollectorIT {
-
- public ColumnEncodedImmutableTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
- }
-
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
- return TestUtil.filterTxParamData(
- Arrays.asList(
- new Object[][] {
- { false, "TEPHRA", false, true },
- { false, "TEPHRA", true, true }, }), 1);
- }
-}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableTxStatsCollectorIT.java
deleted file mode 100644
index 29e000e..0000000
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableTxStatsCollectorIT.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.end2end;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
-import org.apache.phoenix.util.TestUtil;
-import org.junit.runners.Parameterized.Parameters;
-
-public class ColumnEncodedMutableTxStatsCollectorIT extends StatsCollectorIT {
-
- public ColumnEncodedMutableTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
- }
-
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
- return TestUtil.filterTxParamData(
- Arrays.asList(new Object[][] {
- { true, "TEPHRA", false, true },
- { true, "TEPHRA", true, true } }), 1);
- }
-}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableTxStatsCollectorIT.java
deleted file mode 100644
index 04485c4..0000000
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableTxStatsCollectorIT.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.end2end;
-
-import java.util.Arrays;
-import java.util.Collection;
-
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
-import org.apache.phoenix.util.TestUtil;
-import org.junit.runners.Parameterized.Parameters;
-
-public class NonColumnEncodedImmutableTxStatsCollectorIT extends StatsCollectorIT {
-
- public NonColumnEncodedImmutableTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable,transactionProvider, userTableNamespaceMapped, columnEncoded);
- }
-
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
- return TestUtil.filterTxParamData(Arrays.asList(
- new Object[][] {
- { false, "TEPHRA", false, false }, { false, "TEPHRA", true, false },
- { false, "OMID", false, false }, { false, "OMID", true, false },
- }),1);
- }
-}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParallelStatsDisabledIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParallelStatsDisabledIT.java
index 561aee5..8ea8dc8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParallelStatsDisabledIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ParallelStatsDisabledIT.java
@@ -18,8 +18,10 @@
package org.apache.phoenix.end2end;
+import com.google.common.collect.Maps;
import org.apache.commons.lang.StringUtils;
import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.util.QueryBuilder;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
@@ -31,6 +33,7 @@ import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -39,7 +42,13 @@ import static org.junit.Assert.fail;
/**
- * Base class for tests whose methods run in parallel with statistics disabled.
+ * Base class for tests whose methods run in parallel with
+ * 1. Statistics enabled on server side (QueryServices#STATS_COLLECTION_ENABLED is true)
+ * 2. Guide Post Width for all relevant tables is 0. Stats are disabled at table level.
+ *
+ * See {@link org.apache.phoenix.schema.stats.NoOpStatsCollectorIT} for tests that disable
+ * stats collection from server side.
+ *
* You must create unique names using {@link #generateUniqueName()} for each
* table and sequence used to prevent collisions.
*/
@@ -47,8 +56,9 @@ import static org.junit.Assert.fail;
public abstract class ParallelStatsDisabledIT extends BaseTest {
@BeforeClass
- public static final void doSetup() throws Exception {
- setUpTestDriver(ReadOnlyProps.EMPTY_PROPS);
+ public static void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
@AfterClass
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
index 21b2ac9..3ed09c6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/SpillableGroupByIT.java
@@ -76,7 +76,7 @@ public class SpillableGroupByIT extends BaseOwnClusterIT {
// Set guidepost width, but disable stats
props.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(20));
- props.put(QueryServices.STATS_ENABLED_ATTRIB, Boolean.toString(false));
+ props.put(QueryServices.STATS_COLLECTION_ENABLED, Boolean.toString(false));
props.put(QueryServices.EXPLAIN_CHUNK_COUNT_ATTRIB, Boolean.TRUE.toString());
props.put(QueryServices.EXPLAIN_ROW_COUNT_ATTRIB, Boolean.TRUE.toString());
// Must update config before starting server
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/BaseStatsCollectorIT.java
similarity index 84%
rename from phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
rename to phoenix-core/src/it/java/org/apache/phoenix/schema/stats/BaseStatsCollectorIT.java
index 21258e0..40e5c9b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/BaseStatsCollectorIT.java
@@ -19,6 +19,8 @@ package org.apache.phoenix.schema.stats;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_TABLE;
+import static org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MAPREDUCE_JOB_TYPE;
+import static org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MRJobType.UPDATE_STATS;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.apache.phoenix.util.TestUtil.getAllSplits;
import static org.junit.Assert.assertEquals;
@@ -41,13 +43,20 @@ import java.util.Random;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.mapreduce.Job;
import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
import org.apache.phoenix.jdbc.PhoenixConnection;
@@ -63,6 +72,7 @@ import org.apache.phoenix.schema.PTableKey;
import org.apache.phoenix.transaction.PhoenixTransactionProvider.Feature;
import org.apache.phoenix.transaction.TransactionFactory;
import org.apache.phoenix.util.MetaDataUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
@@ -77,8 +87,19 @@ import org.junit.runners.Parameterized;
import com.google.common.collect.Maps;
+/**
+ * Base Test class for all Statistics Collection
+ * Tests stats collection with various scenario parameters
+ * 1. Column Encoding
+ * 2. Transactions
+ * 3. Namespaces
+ * 4. Stats collection via SQL or MR job
+ */
@RunWith(Parameterized.class)
-public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
+public abstract class BaseStatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
+
+ private static final Log LOG = LogFactory.getLog(BaseStatsCollectorIT.class);
+
private final String tableDDLOptions;
private final boolean columnEncoded;
private String tableName;
@@ -89,8 +110,18 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
private final boolean mutable;
private final String transactionProvider;
private static final int defaultGuidePostWidth = 20;
-
- protected StatsCollectorIT(boolean mutable, String transactionProvider, boolean userTableNamespaceMapped, boolean columnEncoded) {
+ private boolean collectStatsOnSnapshot;
+
+ protected BaseStatsCollectorIT(boolean userTableNamespaceMapped, boolean collectStatsOnSnapshot) {
+ this(false, null, userTableNamespaceMapped, false, collectStatsOnSnapshot);
+ }
+
+ protected BaseStatsCollectorIT(boolean mutable, String transactionProvider, boolean columnEncoded) {
+ this(mutable, transactionProvider, false, columnEncoded, false);
+ }
+
+ private BaseStatsCollectorIT(boolean mutable, String transactionProvider,
+ boolean userTableNamespaceMapped, boolean columnEncoded, boolean collectStatsOnSnapshot) {
this.transactionProvider = transactionProvider;
StringBuilder sb = new StringBuilder();
if (columnEncoded) {
@@ -112,16 +143,17 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
this.userTableNamespaceMapped = userTableNamespaceMapped;
this.columnEncoded = columnEncoded;
this.mutable = mutable;
+ this.collectStatsOnSnapshot = collectStatsOnSnapshot;
}
@BeforeClass
public static void doSetup() throws Exception {
- // enable name space mapping at global level on both client and server side
+ // disable name space mapping at global level on both client and server side
Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(7);
- serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.FALSE.toString());
serverProps.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(defaultGuidePostWidth));
Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
- clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.FALSE.toString());
clientProps.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(defaultGuidePostWidth));
setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), new ReadOnlyProps(clientProps.entrySet().iterator()));
}
@@ -148,18 +180,57 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
props.setProperty(QueryServices.EXPLAIN_CHUNK_COUNT_ATTRIB, Boolean.TRUE.toString());
props.setProperty(QueryServices.EXPLAIN_ROW_COUNT_ATTRIB, Boolean.TRUE.toString());
props.setProperty(QueryServices.STATS_UPDATE_FREQ_MS_ATTRIB, Integer.toString(statsUpdateFreq));
- // enable/disable namespace mapping at connection level
props.setProperty(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.toString(userTableNamespaceMapped));
return DriverManager.getConnection(getUrl(), props);
}
-
+
+ private void collectStatistics(Connection conn, String fullTableName) throws Exception {
+ collectStatistics(conn, fullTableName, null);
+ }
+
+ private void collectStatistics(Connection conn, String fullTableName,
+ String guidePostWidth) throws Exception {
+
+ String localPhysicalTableName = SchemaUtil.getPhysicalTableName(fullTableName.getBytes(),
+ userTableNamespaceMapped).getNameAsString();
+
+ if (collectStatsOnSnapshot) {
+ collectStatsOnSnapshot(conn, fullTableName, guidePostWidth, localPhysicalTableName);
+ invalidateStats(conn, fullTableName);
+ } else {
+ String updateStatisticsSql = "UPDATE STATISTICS " + fullTableName;
+ if (guidePostWidth != null) {
+ updateStatisticsSql += " SET \"" + QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB + "\" = " + guidePostWidth;
+ }
+ LOG.info("Running SQL to collect stats: " + updateStatisticsSql);
+ conn.createStatement().execute(updateStatisticsSql);
+ }
+ }
+
+ private void collectStatsOnSnapshot(Connection conn, String fullTableName,
+ String guidePostWidth, String localPhysicalTableName) throws Exception {
+ UpdateStatisticsTool tool = new UpdateStatisticsTool();
+ Configuration conf = utility.getConfiguration();
+ Admin admin = conn.unwrap(PhoenixConnection.class).getQueryServices().getAdmin();
+ String snapshotName = "UpdateStatisticsTool_" + generateUniqueName();
+ admin.snapshot(snapshotName, TableName.valueOf(localPhysicalTableName));
+ LOG.info("Successfully created snapshot " + snapshotName + " for " + localPhysicalTableName);
+ Path randomDir = getUtility().getRandomDir();
+ if (guidePostWidth != null) {
+ conn.createStatement().execute("ALTER TABLE " + fullTableName + " SET GUIDE_POSTS_WIDTH = " + guidePostWidth);
+ }
+ Job job = tool.configureJob(conf, fullTableName, snapshotName, randomDir);
+ assertEquals(job.getConfiguration().get(MAPREDUCE_JOB_TYPE), UPDATE_STATS.name());
+ tool.runJob(job, true);
+ }
+
@Test
public void testUpdateEmptyStats() throws Exception {
Connection conn = getConnection();
conn.setAutoCommit(true);
conn.createStatement().execute(
"CREATE TABLE " + fullTableName +" ( k CHAR(1) PRIMARY KEY )" + tableDDLOptions);
- conn.createStatement().execute("UPDATE STATISTICS " + fullTableName);
+ collectStatistics(conn, fullTableName);
ResultSet rs = conn.createStatement().executeQuery("EXPLAIN SELECT * FROM " + fullTableName);
String explainPlan = QueryUtil.getExplainPlan(rs);
assertEquals(
@@ -168,7 +239,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
explainPlan);
conn.close();
}
-
+
@Test
public void testSomeUpdateEmptyStats() throws Exception {
Connection conn = getConnection();
@@ -176,7 +247,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
conn.createStatement().execute(
"CREATE TABLE " + fullTableName +" ( k VARCHAR PRIMARY KEY, a.v1 VARCHAR, b.v2 VARCHAR ) " + tableDDLOptions + (tableDDLOptions.isEmpty() ? "" : ",") + "SALT_BUCKETS = 3");
conn.createStatement().execute("UPSERT INTO " + fullTableName + "(k,v1) VALUES('a','123456789')");
- conn.createStatement().execute("UPDATE STATISTICS " + fullTableName);
+ collectStatistics(conn, fullTableName);
ResultSet rs;
String explainPlan;
@@ -206,8 +277,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
}
@Test
- public void testUpdateStats() throws SQLException, IOException,
- InterruptedException {
+ public void testUpdateStats() throws Exception {
Connection conn;
PreparedStatement stmt;
ResultSet rs;
@@ -220,9 +290,10 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
String[] s;
Array array;
conn = upsertValues(props, fullTableName);
- // CAll the update statistics query here. If already major compaction has run this will not get executed.
- stmt = conn.prepareStatement("UPDATE STATISTICS " + fullTableName);
- stmt.execute();
+ collectStatistics(conn, fullTableName);
+ rs = conn.createStatement().executeQuery("EXPLAIN SELECT k FROM " + fullTableName);
+ rs.next();
+ long rows1 = (Long) rs.getObject(PhoenixRuntime.EXPLAIN_PLAN_ESTIMATED_ROWS_READ_COLUMN);
stmt = upsertStmt(conn, fullTableName);
stmt.setString(1, "z");
s = new String[] { "xyz", "def", "ghi", "jkll", null, null, "xxx" };
@@ -232,10 +303,12 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
array = conn.createArrayOf("VARCHAR", s);
stmt.setArray(3, array);
stmt.execute();
- stmt = conn.prepareStatement("UPDATE STATISTICS " + fullTableName);
- stmt.execute();
- rs = conn.createStatement().executeQuery("SELECT k FROM " + fullTableName);
- assertTrue(rs.next());
+ conn.commit();
+ collectStatistics(conn, fullTableName);
+ rs = conn.createStatement().executeQuery("EXPLAIN SELECT k FROM " + fullTableName);
+ rs.next();
+ long rows2 = (Long) rs.getObject(PhoenixRuntime.EXPLAIN_PLAN_ESTIMATED_ROWS_READ_COLUMN);
+ assertNotEquals(rows1, rows2);
conn.close();
}
@@ -250,7 +323,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
conn.createStatement().execute("upsert into " + fullTableName + " values ('abc',1,3)");
conn.createStatement().execute("upsert into " + fullTableName + " values ('def',2,4)");
conn.commit();
- conn.createStatement().execute("UPDATE STATISTICS " + fullTableName);
+ collectStatistics(conn, fullTableName);
rs = conn.createStatement().executeQuery("SELECT k FROM " + fullTableName + " order by k desc");
assertTrue(rs.next());
assertEquals("def", rs.getString(1));
@@ -270,57 +343,6 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
testNoDuplicatesAfterUpdateStats(null);
}
- @Test
- public void testUpdateStatsWithMultipleTables() throws Throwable {
- String fullTableName2 = SchemaUtil.getTableName(schemaName, "T_" + generateUniqueName());
- Connection conn;
- PreparedStatement stmt;
- ResultSet rs;
- Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
- conn = getConnection();
- conn.createStatement().execute(
- "CREATE TABLE " + fullTableName +" ( k VARCHAR, a_string_array VARCHAR(100) ARRAY[4], b_string_array VARCHAR(100) ARRAY[4] \n"
- + " CONSTRAINT pk PRIMARY KEY (k, b_string_array DESC))" + tableDDLOptions );
- conn.createStatement().execute(
- "CREATE TABLE " + fullTableName2 +" ( k VARCHAR, a_string_array VARCHAR(100) ARRAY[4], b_string_array VARCHAR(100) ARRAY[4] \n"
- + " CONSTRAINT pk PRIMARY KEY (k, b_string_array DESC))" + tableDDLOptions );
- String[] s;
- Array array;
- conn = upsertValues(props, fullTableName);
- conn = upsertValues(props, fullTableName2);
- // CAll the update statistics query here
- stmt = conn.prepareStatement("UPDATE STATISTICS "+fullTableName);
- stmt.execute();
- stmt = conn.prepareStatement("UPDATE STATISTICS "+fullTableName2);
- stmt.execute();
- stmt = upsertStmt(conn, fullTableName);
- stmt.setString(1, "z");
- s = new String[] { "xyz", "def", "ghi", "jkll", null, null, "xxx" };
- array = conn.createArrayOf("VARCHAR", s);
- stmt.setArray(2, array);
- s = new String[] { "zya", "def", "ghi", "jkll", null, null, null, "xxx" };
- array = conn.createArrayOf("VARCHAR", s);
- stmt.setArray(3, array);
- stmt.execute();
- stmt = upsertStmt(conn, fullTableName2);
- stmt.setString(1, "z");
- s = new String[] { "xyz", "def", "ghi", "jkll", null, null, "xxx" };
- array = conn.createArrayOf("VARCHAR", s);
- stmt.setArray(2, array);
- s = new String[] { "zya", "def", "ghi", "jkll", null, null, null, "xxx" };
- array = conn.createArrayOf("VARCHAR", s);
- stmt.setArray(3, array);
- stmt.execute();
- conn.close();
- conn = getConnection();
- // This analyze would not work
- stmt = conn.prepareStatement("UPDATE STATISTICS "+fullTableName2);
- stmt.execute();
- rs = conn.createStatement().executeQuery("SELECT k FROM "+fullTableName2);
- assertTrue(rs.next());
- conn.close();
- }
-
private Connection upsertValues(Properties props, String tableName) throws SQLException, IOException,
InterruptedException {
Connection conn;
@@ -395,10 +417,6 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
return stmt;
}
- private void compactTable(Connection conn, String tableName) throws Exception {
- TestUtil.doMajorCompaction(conn, tableName);
- }
-
@Test
@Ignore //TODO remove this once https://issues.apache.org/jira/browse/TEPHRA-208 is fixed
public void testCompactUpdatesStats() throws Exception {
@@ -434,8 +452,8 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
stmt.executeUpdate();
}
conn.commit();
-
- compactTable(conn, physicalTableName);
+
+ TestUtil.doMajorCompaction(conn, physicalTableName);
if (statsUpdateFreq != 0) {
invalidateStats(conn, tableName);
@@ -466,8 +484,8 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
System.out.println(result);
}
}
-
- compactTable(conn, physicalTableName);
+
+ TestUtil.doMajorCompaction(conn, physicalTableName);
scan = new Scan();
scan.setRaw(true);
@@ -527,7 +545,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
conn.commit();
ResultSet rs;
- TestUtil.analyzeTable(conn, fullTableName);
+ collectStatistics(conn, fullTableName);
List<KeyRange> keyRanges = getAllSplits(conn, fullTableName);
assertEquals(26, keyRanges.size());
rs = conn.createStatement().executeQuery("EXPLAIN SELECT * FROM " + fullTableName);
@@ -538,10 +556,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
List<HRegionLocation> regions = services.getAllTableRegions(Bytes.toBytes(physicalTableName));
assertEquals(1, regions.size());
- TestUtil.analyzeTable(conn, fullTableName);
- String query = "UPDATE STATISTICS " + fullTableName + " SET \""
- + QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB + "\"=" + Long.toString(1000);
- conn.createStatement().execute(query);
+ collectStatistics(conn, fullTableName, Long.toString(1000));
keyRanges = getAllSplits(conn, fullTableName);
boolean oneCellPerColFamliyStorageScheme = !mutable && columnEncoded;
boolean hasShadowCells = TransactionFactory.Provider.OMID.name().equals(transactionProvider);
@@ -582,7 +597,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
// Disable stats
conn.createStatement().execute("ALTER TABLE " + fullTableName +
" SET " + PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH + "=0");
- TestUtil.analyzeTable(conn, fullTableName);
+ collectStatistics(conn, fullTableName);
// Assert that there are no more guideposts
rs = conn.createStatement().executeQuery("SELECT count(1) FROM " + PhoenixDatabaseMetaData.SYSTEM_STATS_NAME +
" WHERE " + PhoenixDatabaseMetaData.PHYSICAL_NAME + "='" + physicalTableName + "' AND " + PhoenixDatabaseMetaData.COLUMN_FAMILY + " IS NOT NULL");
@@ -595,7 +610,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
}
@Test
- public void testRowCountAndByteCounts() throws SQLException {
+ public void testRowCountAndByteCounts() throws Exception {
Connection conn = getConnection();
String ddl = "CREATE TABLE " + fullTableName + " (t_id VARCHAR NOT NULL,\n" + "k1 INTEGER NOT NULL,\n"
+ "k2 INTEGER NOT NULL,\n" + "C3.k3 INTEGER,\n" + "C2.v1 VARCHAR,\n"
@@ -610,9 +625,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
}
conn.commit();
ResultSet rs;
- String query = "UPDATE STATISTICS " + fullTableName + " SET \""
- + QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB + "\"=" + Long.toString(20);
- conn.createStatement().execute(query);
+ collectStatistics(conn, fullTableName, Long.toString(20L));
Random r = new Random();
int count = 0;
boolean hasShadowCells = TransactionFactory.Provider.OMID.name().equals(transactionProvider);
@@ -643,11 +656,10 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
@Test
public void testRowCountWhenNumKVsExceedCompactionScannerThreshold() throws Exception {
- String tableName = generateUniqueName();
StringBuilder sb = new StringBuilder(200);
- sb.append("CREATE TABLE " + tableName + "(PK1 VARCHAR NOT NULL, ");
+ sb.append("CREATE TABLE " + fullTableName + "(PK1 VARCHAR NOT NULL, ");
int numRows = 10;
- try (Connection conn = DriverManager.getConnection(getUrl())) {
+ try (Connection conn = getConnection()) {
int compactionScannerKVThreshold =
conn.unwrap(PhoenixConnection.class).getQueryServices().getConfiguration()
.getInt(HConstants.COMPACTION_KV_MAX,
@@ -663,7 +675,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
String ddl = sb.toString();
conn.createStatement().execute(ddl);
sb = new StringBuilder(200);
- sb.append("UPSERT INTO " + tableName + " VALUES (");
+ sb.append("UPSERT INTO " + fullTableName + " VALUES (");
for (int i = 1; i <= numKvColumns + 1; i++) {
sb.append("?");
if (i < numKvColumns + 1) {
@@ -685,15 +697,15 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
stmt.executeUpdate();
}
conn.commit();
- conn.createStatement().execute("UPDATE STATISTICS " + tableName);
- String q = "SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + tableName + "'";
+ collectStatistics(conn, fullTableName);
+ String q = "SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + physicalTableName + "'";
ResultSet rs = conn.createStatement().executeQuery(q);
rs.next();
assertEquals("Number of expected rows in stats table after update stats didn't match!", numRows, rs.getInt(1));
- conn.createStatement().executeUpdate("DELETE FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + tableName + "'");
+ conn.createStatement().executeUpdate("DELETE FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + physicalTableName + "'");
conn.commit();
- TestUtil.doMajorCompaction(conn, tableName);
- q = "SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + tableName + "'";
+ TestUtil.doMajorCompaction(conn, physicalTableName);
+ q = "SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + physicalTableName + "'";
rs = conn.createStatement().executeQuery(q);
rs.next();
assertEquals("Number of expected rows in stats table after major compaction didn't match", numRows, rs.getInt(1));
@@ -721,52 +733,49 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
@Test
public void testEmptyGuidePostGeneratedWhenDataSizeLessThanGPWidth() throws Exception {
- String tableName = generateUniqueName();
- try (Connection conn = DriverManager.getConnection(getUrl())) {
+ try (Connection conn = getConnection()) {
long guidePostWidth = 20000000;
conn.createStatement()
- .execute("CREATE TABLE " + tableName
+ .execute("CREATE TABLE " + fullTableName
+ " ( k INTEGER, c1.a bigint,c2.b bigint CONSTRAINT pk PRIMARY KEY (k)) GUIDE_POSTS_WIDTH="
+ guidePostWidth + ", SALT_BUCKETS = 4");
- conn.createStatement().execute("upsert into " + tableName + " values (100,1,3)");
- conn.createStatement().execute("upsert into " + tableName + " values (101,2,4)");
+ conn.createStatement().execute("upsert into " + fullTableName + " values (100,1,3)");
+ conn.createStatement().execute("upsert into " + fullTableName + " values (101,2,4)");
conn.commit();
- conn.createStatement().execute("UPDATE STATISTICS " + tableName);
+ collectStatistics(conn, fullTableName);
ConnectionQueryServices queryServices =
conn.unwrap(PhoenixConnection.class).getQueryServices();
- verifyGuidePostGenerated(queryServices, tableName, new String[] {"C1", "C2"}, guidePostWidth, true);
+ verifyGuidePostGenerated(queryServices, physicalTableName, new String[] {"C1", "C2"}, guidePostWidth, true);
}
}
@Test
public void testCollectingAllVersionsOfCells() throws Exception {
- String tableName = generateUniqueName();
- try (Connection conn = DriverManager.getConnection(getUrl())) {
+ try (Connection conn = getConnection()) {
long guidePostWidth = 70;
String ddl =
- "CREATE TABLE " + tableName + " (k INTEGER PRIMARY KEY, c1.a bigint, c2.b bigint)"
+ "CREATE TABLE " + fullTableName + " (k INTEGER PRIMARY KEY, c1.a bigint, c2.b bigint)"
+ " GUIDE_POSTS_WIDTH=" + guidePostWidth
+ ", USE_STATS_FOR_PARALLELIZATION=true" + ", VERSIONS=3";
conn.createStatement().execute(ddl);
- conn.createStatement().execute("upsert into " + tableName + " values (100,100,3)");
+ conn.createStatement().execute("upsert into " + fullTableName + " values (100,100,3)");
conn.commit();
- conn.createStatement().execute("UPDATE STATISTICS " + tableName);
+ collectStatistics(conn, fullTableName);
ConnectionQueryServices queryServices =
conn.unwrap(PhoenixConnection.class).getQueryServices();
// The table only has one row. All cells just has one version, and the data size of the row
// is less than the guide post width, so we generate empty guide post.
- verifyGuidePostGenerated(queryServices, tableName, new String[] {"C1", "C2"}, guidePostWidth, true);
-
+ verifyGuidePostGenerated(queryServices, physicalTableName, new String[] {"C1", "C2"}, guidePostWidth, true);
- conn.createStatement().execute("upsert into " + tableName + " values (100,101,4)");
+ conn.createStatement().execute("upsert into " + fullTableName + " values (100,101,4)");
conn.commit();
- conn.createStatement().execute("UPDATE STATISTICS " + tableName);
+ collectStatistics(conn, fullTableName);
// We updated the row. Now each cell has two versions, and the data size of the row
// is >= the guide post width, so we generate non-empty guide post.
- verifyGuidePostGenerated(queryServices, tableName, new String[] {"C1", "C2"}, guidePostWidth, false);
+ verifyGuidePostGenerated(queryServices, physicalTableName, new String[] {"C1", "C2"}, guidePostWidth, false);
}
}
@@ -836,7 +845,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
conn.createStatement()
.execute("ALTER TABLE " + baseTable + " SET GUIDE_POSTS_WIDTH=" + newGpWidth);
- // base table
+ // base table and local index
statsCollector = getDefaultStatsCollectorForTable(baseTable);
statsCollector.init();
assertEquals(newGpWidth, statsCollector.getGuidePostDepth());
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableNonTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceDisabledStatsCollectorIT.java
similarity index 53%
rename from phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableNonTxStatsCollectorIT.java
rename to phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceDisabledStatsCollectorIT.java
index 248f6fb..138c7c0 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedMutableNonTxStatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceDisabledStatsCollectorIT.java
@@ -15,24 +15,30 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.phoenix.end2end;
+
+package org.apache.phoenix.schema.stats;
+
+import org.junit.runners.Parameterized;
import java.util.Arrays;
import java.util.Collection;
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
-import org.junit.runners.Parameterized.Parameters;
+public class NamespaceDisabledStatsCollectorIT extends BaseStatsCollectorIT {
-public class ColumnEncodedMutableNonTxStatsCollectorIT extends StatsCollectorIT {
-
- public ColumnEncodedMutableNonTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
+ public NamespaceDisabledStatsCollectorIT(boolean userTableNamespaceMapped, boolean collectStatsOnSnapshot) {
+ super(userTableNamespaceMapped, collectStatsOnSnapshot);
}
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
+ @Parameterized.Parameters(name = "userTableNamespaceMapped={0},collectStatsOnSnapshot={1}")
+ public static Collection<Object[]> provideData() {
return Arrays.asList(
- new Object[][] { { true, null, false, true }, { true, null, true, true } });
+ new Object[][] {
+ // Collect stats on snapshots using UpdateStatisticsTool
+ { false, true },
+ // Collect stats via `UPDATE STATISTICS` SQL
+ { false, false }
+ }
+ );
}
+
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SysTableNamespaceMappedStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceEnabledStatsCollectorIT.java
similarity index 60%
rename from phoenix-core/src/it/java/org/apache/phoenix/end2end/SysTableNamespaceMappedStatsCollectorIT.java
rename to phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceEnabledStatsCollectorIT.java
index 8f0a1d8..d6b1f4e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/SysTableNamespaceMappedStatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NamespaceEnabledStatsCollectorIT.java
@@ -15,50 +15,47 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.phoenix.end2end;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Map;
+package org.apache.phoenix.schema.stats;
+import com.google.common.collect.Maps;
import org.apache.phoenix.query.QueryServices;
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
import org.apache.phoenix.util.ReadOnlyProps;
-import org.apache.phoenix.util.TestUtil;
import org.junit.BeforeClass;
-import org.junit.runners.Parameterized.Parameters;
+import org.junit.runners.Parameterized;
-import com.google.common.collect.Maps;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
-public class SysTableNamespaceMappedStatsCollectorIT extends StatsCollectorIT {
+public class NamespaceEnabledStatsCollectorIT extends BaseStatsCollectorIT {
- public SysTableNamespaceMappedStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
+ public NamespaceEnabledStatsCollectorIT(boolean userTableNamespaceMapped, boolean collectStatsOnSnapshot) {
+ super(userTableNamespaceMapped, collectStatsOnSnapshot);
}
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
- return TestUtil.filterTxParamData(Arrays.asList(
- new Object[][] {
- { true, "TEPHRA", false, false }, { true, "TEPHRA", false, true },
- { true, "OMID", false, false },
- }), 1);
+ @Parameterized.Parameters(name = "userTableNamespaceMapped={0},collectStatsOnSnapshot={1}")
+ public static Collection<Object[]> provideData() {
+ return Arrays.asList(
+ new Object[][] {
+ // Collect stats on snapshots using UpdateStatisticsTool
+ { true, true },
+ // Collect stats via `UPDATE STATISTICS` SQL
+ { true, false }
+ }
+ );
}
@BeforeClass
public static void doSetup() throws Exception {
// enable name space mapping at global level on both client and server side
Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(7);
- serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ serverProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.TRUE.toString());
serverProps.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(20));
- serverProps.put(QueryServices.IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, "true");
Map<String, String> clientProps = Maps.newHashMapWithExpectedSize(2);
- clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, "true");
+ clientProps.put(QueryServices.IS_NAMESPACE_MAPPING_ENABLED, Boolean.TRUE.toString());
clientProps.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(20));
- clientProps.put(QueryServices.IS_SYSTEM_TABLE_MAPPED_TO_NAMESPACE, "true");
setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()),
- new ReadOnlyProps(clientProps.entrySet().iterator()));
+ new ReadOnlyProps(clientProps.entrySet().iterator()));
}
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NoOpStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NoOpStatsCollectorIT.java
new file mode 100644
index 0000000..04f4143
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NoOpStatsCollectorIT.java
@@ -0,0 +1,143 @@
+/*
+ * 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.schema.stats;
+
+import com.google.common.collect.Maps;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.sql.Array;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Map;
+import java.util.Properties;
+
+import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+
+/**
+ * Tests the behavior of stats collection code when stats are disabled on server side
+ * explicitly using QueryServices#STATS_COLLECTION_ENABLED property
+ */
+public class NoOpStatsCollectorIT extends ParallelStatsDisabledIT {
+
+ private static final Log LOG = LogFactory.getLog(NoOpStatsCollectorIT.class);
+
+ private String fullTableName;
+ private String physicalTableName;
+ private Connection conn;
+
+ /**
+ * Disable stats on server side by setting QueryServices#STATS_COLLECTION_ENABLED to false
+ */
+ @BeforeClass
+ public static void doSetup() throws Exception {
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(1);
+ props.put(QueryServices.STATS_COLLECTION_ENABLED, Boolean.FALSE.toString());
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Before
+ public void beforeTest() throws SQLException {
+ String schemaName = generateUniqueName();
+ String tableName = "T_" + generateUniqueName();
+ fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+ physicalTableName = SchemaUtil.getPhysicalHBaseTableName(schemaName,
+ tableName, false).getString();
+ conn = getConnection();
+ conn.createStatement().execute(
+ "CREATE TABLE " + fullTableName + " ( k VARCHAR, a_string_array VARCHAR(100) ARRAY[4],"
+ + " b_string_array VARCHAR(100) ARRAY[4] \n"
+ + " CONSTRAINT pk PRIMARY KEY (k, b_string_array DESC)) GUIDE_POSTS_WIDTH = 10");
+ upsertValues(conn, fullTableName);
+ }
+
+ /**
+ * Update Statistics SQL statement should be disallowed
+ */
+ @Test
+ public void testStatsCollectionViaSql() throws SQLException {
+ String updateStatisticsSql = "UPDATE STATISTICS " + fullTableName;
+ LOG.info("Running SQL to collect stats: " + updateStatisticsSql);
+ Statement stmt = conn.createStatement();
+ try {
+ stmt.execute(updateStatisticsSql);
+ } catch (SQLException e) {
+ Assert.assertEquals("StatsCollectionDisabledOnServerException expected",
+ 1401, e.getErrorCode());
+ Assert.assertEquals("StatsCollectionDisabledOnServerException expected",
+ "STS01", e.getSQLState());
+ }
+ }
+
+ /**
+ * Major compaction should not compute / persist statistics
+ */
+ @Test
+ public void testStatsCollectionDuringMajorCompaction() throws Exception {
+ LOG.info("Running major compaction on table: " + physicalTableName);
+ TestUtil.doMajorCompaction(conn, physicalTableName);
+
+ String q1 = "SELECT SUM(GUIDE_POSTS_ROW_COUNT) FROM SYSTEM.STATS WHERE PHYSICAL_NAME = '" + physicalTableName + "'";
+ ResultSet rs1 = conn.createStatement().executeQuery(q1);
+ rs1.next();
+ int rowCountFromStats = rs1.getInt(1);
+
+ String q2 = "SELECT COUNT(*) FROM " + fullTableName;
+ ResultSet rs2 = conn.createStatement().executeQuery(q2);
+ rs2.next();
+ int rowCountFromTable = rs2.getInt(1);
+
+ Assert.assertTrue("Stats collection is disabled, hence row counts should not match",
+ rowCountFromStats != rowCountFromTable);
+ Assert.assertEquals("Stats collection is disabled, hence row counts from stats should be 0",
+ 0, rowCountFromStats);
+ }
+
+ private Connection getConnection() throws SQLException {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ return DriverManager.getConnection(getUrl(), props);
+ }
+
+ private void upsertValues(Connection conn, String tableName) throws SQLException {
+ PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " VALUES(?,?,?)");
+ stmt.setString(1, "a");
+ String[] s = new String[] { "abc", "def", "ghi", "jkll", null, null, "xxx" };
+ Array array = conn.createArrayOf("VARCHAR", s);
+ stmt.setArray(2, array);
+ s = new String[] { "abc", "def", "ghi", "jkll", null, null, null, "xxx" };
+ array = conn.createArrayOf("VARCHAR", s);
+ stmt.setArray(3, array);
+ stmt.execute();
+ conn.commit();
+ }
+
+}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableNonTxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NonTxStatsCollectorIT.java
similarity index 55%
rename from phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableNonTxStatsCollectorIT.java
rename to phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NonTxStatsCollectorIT.java
index fec890e..651a21b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/NonColumnEncodedImmutableNonTxStatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/NonTxStatsCollectorIT.java
@@ -15,24 +15,32 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.phoenix.end2end;
+package org.apache.phoenix.schema.stats;
import java.util.Arrays;
import java.util.Collection;
-import org.apache.phoenix.schema.stats.StatsCollectorIT;
import org.junit.runners.Parameterized.Parameters;
-public class NonColumnEncodedImmutableNonTxStatsCollectorIT extends StatsCollectorIT {
+public class NonTxStatsCollectorIT extends BaseStatsCollectorIT {
- public NonColumnEncodedImmutableNonTxStatsCollectorIT(boolean mutable, String transactionProvider,
- boolean userTableNamespaceMapped, boolean columnEncoded) {
- super(mutable, transactionProvider, userTableNamespaceMapped, columnEncoded);
+ public NonTxStatsCollectorIT(boolean mutable,
+ String transactionProvider, boolean columnEncoded) {
+ super(mutable, transactionProvider, columnEncoded);
}
- @Parameters(name = "mutable={0},transactionProvider={1},isUserTableNamespaceMapped={2},columnEncoded={3}")
- public static Collection<Object[]> data() {
+ @Parameters(name = "mutable={0},transactionProvider={1},columnEncoded={2}")
+ public static Collection<Object[]> provideData() {
return Arrays.asList(
- new Object[][] { { false, null, false, false }, { false, null, true, false } });
+ new Object[][] {
+ // Immutable, Column Encoded
+ { false, null, true },
+ // Mutable, Column Encoded
+ { true, null, true },
+ // Immutable, Not Column Encoded
+ { false, null, false }
+ }
+ );
}
+
}
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/TxStatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/TxStatsCollectorIT.java
new file mode 100644
index 0000000..fc26c3e
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/TxStatsCollectorIT.java
@@ -0,0 +1,51 @@
+/*
+ * 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.schema.stats;
+
+import org.apache.phoenix.util.TestUtil;
+import org.junit.runners.Parameterized;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+public class TxStatsCollectorIT extends BaseStatsCollectorIT {
+
+ public TxStatsCollectorIT(boolean mutable, String transactionProvider, boolean columnEncoded) {
+ super(mutable, transactionProvider, columnEncoded);
+ }
+
+ @Parameterized.Parameters(name = "mutable={0},transactionProvider={1},columnEncoded={2}")
+ public static Collection<Object[]> data() {
+ return TestUtil.filterTxParamData(
+ Arrays.asList(
+ new Object[][] {
+ // Immutable, TEPHRA, Column Encoded
+ { false, "TEPHRA", true },
+ // Immutable, TEPHRA, Non Column Encoded
+ { false, "TEPHRA", false },
+ // Immutable, OMID, Non Column Encoded
+ { false, "OMID", false },
+
+ // Mutable, TEPHRA, Column Encoded
+ { true, "TEPHRA", true },
+ // Mutable, TEPHRA, Non Column Encoded
+ { true, "TEPHRA", false },
+ // Mutable, OMID, Non Column Encoded
+ { true, "OMID", false }}), 1);
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 2fc31c0..36717fb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -123,9 +123,12 @@ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableNotFoundException;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.ValueSchema.Field;
+import org.apache.phoenix.schema.stats.NoOpStatisticsCollector;
import org.apache.phoenix.schema.stats.StatisticsCollectionRunTracker;
import org.apache.phoenix.schema.stats.StatisticsCollector;
import org.apache.phoenix.schema.stats.StatisticsCollectorFactory;
+import org.apache.phoenix.schema.stats.StatisticsScanner;
+import org.apache.phoenix.schema.stats.StatsCollectionDisabledOnServerException;
import org.apache.phoenix.schema.tuple.EncodedColumnQualiferCellsList;
import org.apache.phoenix.schema.tuple.MultiKeyValueTuple;
import org.apache.phoenix.schema.tuple.PositionBasedMultiKeyValueTuple;
@@ -397,7 +400,11 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
StatisticsCollector statsCollector = StatisticsCollectorFactory.createStatisticsCollector(
env, region.getRegionInfo().getTable().getNameAsString(), ts,
gp_width_bytes, gp_per_region_bytes);
- return collectStats(s, statsCollector, region, scan, env.getConfiguration());
+ if (statsCollector instanceof NoOpStatisticsCollector) {
+ throw new StatsCollectionDisabledOnServerException();
+ } else {
+ return collectStats(s, statsCollector, region, scan, env.getConfiguration());
+ }
} else if (ScanUtil.isIndexRebuild(scan)) {
return rebuildIndices(s, region, scan, env.getConfiguration());
}
@@ -1004,32 +1011,38 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
CompactionRequest request) throws IOException {
if (scanType.equals(ScanType.COMPACT_DROP_DELETES)) {
final TableName table = c.getEnvironment().getRegion().getRegionInfo().getTable();
- // Compaction and split upcalls run with the effective user context of the requesting user.
- // This will lead to failure of cross cluster RPC if the effective user is not
- // the login user. Switch to the login user context to ensure we have the expected
- // security context.
- return User.runAsLoginUser(new PrivilegedExceptionAction<InternalScanner>() {
- @Override public InternalScanner run() throws Exception {
- InternalScanner internalScanner = scanner;
- try {
- long clientTimeStamp = EnvironmentEdgeManager.currentTimeMillis();
- DelegateRegionCoprocessorEnvironment compactionConfEnv = new DelegateRegionCoprocessorEnvironment(c.getEnvironment(), ConnectionType.COMPACTION_CONNECTION);
- StatisticsCollector stats = StatisticsCollectorFactory.createStatisticsCollector(
- compactionConfEnv, table.getNameAsString(), clientTimeStamp,
- store.getColumnFamilyDescriptor().getName());
- internalScanner =
- stats.createCompactionScanner(compactionConfEnv,
- store, scanner);
- } catch (Exception e) {
- // If we can't reach the stats table, don't interrupt the normal
- // compaction operation, just log a warning.
- if (logger.isWarnEnabled()) {
- logger.warn("Unable to collect stats for " + table, e);
- }
- }
- return internalScanner;
+ // Compaction and split upcalls run with the effective user context of the requesting user.
+ // This will lead to failure of cross cluster RPC if the effective user is not
+ // the login user. Switch to the login user context to ensure we have the expected
+ // security context.
+ return User.runAsLoginUser(
+ new PrivilegedExceptionAction<InternalScanner>() {
+ @Override
+ public InternalScanner run() throws Exception {
+ InternalScanner internalScanner = scanner;
+ try {
+ long clientTimeStamp = EnvironmentEdgeManager.currentTimeMillis();
+ DelegateRegionCoprocessorEnvironment compactionConfEnv =
+ new DelegateRegionCoprocessorEnvironment(
+ c.getEnvironment(), ConnectionType.COMPACTION_CONNECTION);
+ StatisticsCollector statisticsCollector =
+ StatisticsCollectorFactory.createStatisticsCollector(
+ compactionConfEnv,
+ table.getNameAsString(),
+ clientTimeStamp,
+ store.getColumnFamilyDescriptor().getName());
+ statisticsCollector.init();
+ internalScanner = statisticsCollector.createCompactionScanner(compactionConfEnv, store, scanner);
+ } catch (Exception e) {
+ // If we can't reach the stats table, don't interrupt the normal
+ // compaction operation, just log a warning.
+ if (logger.isWarnEnabled()) {
+ logger.warn("Unable to collect stats for " + table, e);
}
- });
+ }
+ return internalScanner;
+ }
+ });
}
return scanner;
}
@@ -1270,7 +1283,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
} finally {
try {
if (noErrors && !compactionRunning) {
- statsCollector.updateStatistic(region, scan);
+ statsCollector.updateStatistics(region, scan);
logger.info("UPDATE STATISTICS finished successfully for scanner: "
+ innerScanner + ". Number of rows scanned: " + rowCount
+ ". Time: " + (System.currentTimeMillis() - startTime));
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 2c52d44..e531529 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -464,6 +464,8 @@ public enum SQLExceptionCode {
INSUFFICIENT_MEMORY(999, "50M01", "Unable to allocate enough memory."),
HASH_JOIN_CACHE_NOT_FOUND(900, "HJ01", "Hash Join cache not found"),
+ STATS_COLLECTION_DISABLED_ON_SERVER(1401, "STS01", "Stats collection attempted but is disabled on server"),
+
CANNOT_UPSERT_WITH_SCN_FOR_ROW_TIMSTAMP_COLUMN(901,"43M12",
"Cannot use a connection with SCN set to upsert data for " +
"table with ROW_TIMESTAMP column."),
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
index 5065300..c25b516 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SnapshotScanner.java
@@ -18,12 +18,10 @@
package org.apache.phoenix.iterate;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -47,35 +45,65 @@ import org.apache.hadoop.hbase.regionserver.OnlineRegions;
import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.RegionScanner;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
-import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.util.EncodedColumnsUtil;
-
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.mapreduce.util.ConnectionUtil;
+import org.apache.phoenix.schema.stats.DefaultStatisticsCollector;
+import org.apache.phoenix.schema.stats.NoOpStatisticsCollector;
+import org.apache.phoenix.schema.stats.StatisticsCollector;
+import org.apache.phoenix.schema.stats.StatisticsWriter;
+import org.apache.phoenix.util.ScanUtil;
+import org.apache.phoenix.util.SchemaUtil;
+
+import java.util.Properties;
+
+/**
+ * Scan over a region from restored snapshot
+ */
public class SnapshotScanner extends AbstractClientScanner {
private static final Log LOG = LogFactory.getLog(SnapshotScanner.class);
-
- private RegionScanner scanner = null;
+ private final Scan scan;
+ private RegionScanner scanner;
private HRegion region;
- List<Cell> values;
+ private List<Cell> values;
+ private StatisticsCollector statisticsCollector;
public SnapshotScanner(Configuration conf, FileSystem fs, Path rootDir,
TableDescriptor htd, RegionInfo hri, Scan scan) throws Throwable{
+ LOG.info("Creating SnapshotScanner for region: " + hri);
+
scan.setIsolationLevel(IsolationLevel.READ_UNCOMMITTED);
values = new ArrayList<>();
this.region = HRegion.openHRegion(conf, fs, rootDir, hri, htd, null, null, null);
+ this.scan = scan;
RegionCoprocessorEnvironment snapshotEnv = getSnapshotContextEnvironment(conf);
- RegionScannerFactory regionScannerFactory;
- if (scan.getAttribute(BaseScannerRegionObserver.NON_AGGREGATE_QUERY) != null) {
- regionScannerFactory = new NonAggregateRegionScannerFactory(snapshotEnv);
+ // Collect statistics during scan if ANALYZE_TABLE attribute is set
+ if (ScanUtil.isAnalyzeTable(scan)) {
+ this.scanner = region.getScanner(scan);
+ PhoenixConnection connection = (PhoenixConnection) ConnectionUtil.getInputConnection(conf, new Properties());
+ String tableName = region.getTableDescriptor().getTableName().getNameAsString();
+ TableName physicalTableName = SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, conf);
+ Table table = connection.getQueryServices().getTable(physicalTableName.getName());
+ StatisticsWriter statsWriter = StatisticsWriter.newWriter(connection, tableName, HConstants.LATEST_TIMESTAMP);
+ statisticsCollector = new DefaultStatisticsCollector(conf, region,
+ tableName, null, null, null, statsWriter, table);
+ } else if (scan.getAttribute(BaseScannerRegionObserver.NON_AGGREGATE_QUERY) != null) {
+ RegionScannerFactory regionScannerFactory = new NonAggregateRegionScannerFactory(snapshotEnv);
+ this.scanner = regionScannerFactory.getRegionScanner(scan, region.getScanner(scan));
+ statisticsCollector = new NoOpStatisticsCollector();
} else {
/* future work : Snapshot M/R jobs for aggregate queries*/
throw new UnsupportedOperationException("Snapshot M/R jobs not available for aggregate queries");
}
- this.scanner = regionScannerFactory.getRegionScanner(scan, region.getScanner(scan));
+ statisticsCollector.init();
region.startRegionOperation();
}
@@ -84,6 +112,7 @@ public class SnapshotScanner extends AbstractClientScanner {
public Result next() throws IOException {
values.clear();
scanner.nextRaw(values);
+ statisticsCollector.collectStatistics(values);
if (values.isEmpty()) {
//we are done
return null;
@@ -96,6 +125,7 @@ public class SnapshotScanner extends AbstractClientScanner {
public void close() {
if (this.scanner != null) {
try {
+ statisticsCollector.updateStatistics(region, scan);
this.scanner.close();
this.scanner = null;
} catch (IOException e) {
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
index 1d490c1..99d71b5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableSnapshotResultIterator.java
@@ -21,15 +21,14 @@ package org.apache.phoenix.iterate;
import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
import java.util.List;
import java.util.UUID;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.PrivateCellUtil;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.Scan;
@@ -42,15 +41,26 @@ import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.ServerUtil;
+
+/**
+ * Iterator to scan over a HBase snapshot based on input HBase Scan object.
+ * This iterator is generated by Phoenix during the query plan scan generation,
+ * hence it will include scan attributes and custom filters.
+ * Restores HBase snapshot and determines the valid regions that intersect
+ * with the input Scan boundaries. Launches SnapshotScanner for each of them.
+ * Deletes the restored snapshot when iterator is closed.
+ */
public class TableSnapshotResultIterator implements ResultIterator {
+ private static final Log LOG = LogFactory.getLog(TableSnapshotResultIterator.class);
+
private final Scan scan;
private ResultIterator scanIterator;
private Configuration configuration;
private final ScanMetricsHolder scanMetricsHolder;
private Tuple lastTuple = null;
private static final ResultIterator UNINITIALIZED_SCANNER = ResultIterator.EMPTY_ITERATOR;
- private ArrayList<RegionInfo> regions;
+ private List<RegionInfo> regions;
private TableDescriptor htd;
private String snapshotName;
@@ -82,7 +92,7 @@ public class TableSnapshotResultIterator implements ResultIterator {
this.rootDir, this.restoreDir, this.snapshotName);
List<RegionInfo> restoredRegions = meta.getRegionsToAdd();
this.htd = meta.getTableDescriptor();
- this.regions = new ArrayList<>(restoredRegions.size());
+ this.regions = new ArrayList<RegionInfo>(restoredRegions.size());
for (RegionInfo restoredRegion : restoredRegions) {
if (isValidRegion(restoredRegion)) {
@@ -91,6 +101,7 @@ public class TableSnapshotResultIterator implements ResultIterator {
}
this.regions.sort(RegionInfo.COMPARATOR);
+ LOG.info("Initialization complete with " + regions.size() + " valid regions");
}
/**
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java
index 4186c81..6f83800 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixInputFormat.java
@@ -17,14 +17,8 @@
*/
package org.apache.phoenix.mapreduce;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.Collections;
-import java.util.List;
-import java.util.Properties;
-
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -47,12 +41,19 @@ import org.apache.phoenix.iterate.MapReduceParallelScanGrouper;
import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.mapreduce.util.ConnectionUtil;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MRJobType;
import org.apache.phoenix.query.HBaseFactoryProvider;
import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.stats.StatisticsUtil;
import org.apache.phoenix.util.PhoenixRuntime;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
/**
* {@link InputFormat} implementation from Phoenix.
@@ -78,8 +79,6 @@ public class PhoenixInputFormat<T extends DBWritable> extends InputFormat<NullWr
final Class<T> inputClass = (Class<T>) PhoenixConfigurationUtil.getInputClass(configuration);
return new PhoenixRecordReader<T>(inputClass , configuration, queryPlan);
}
-
-
@Override
public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
@@ -163,8 +162,7 @@ public class PhoenixInputFormat<T extends DBWritable> extends InputFormat<NullWr
* @throws IOException
* @throws SQLException
*/
- private QueryPlan getQueryPlan(final JobContext context, final Configuration configuration)
- throws IOException {
+ private QueryPlan getQueryPlan(final JobContext context, final Configuration configuration) {
Preconditions.checkNotNull(context);
try {
final String txnScnValue = configuration.get(PhoenixConfigurationUtil.TX_SCN_VALUE);
@@ -180,13 +178,28 @@ public class PhoenixInputFormat<T extends DBWritable> extends InputFormat<NullWr
try (final Connection connection = ConnectionUtil.getInputConnection(configuration, overridingProps);
final Statement statement = connection.createStatement()) {
- final String selectStatement = PhoenixConfigurationUtil.getSelectStatement(configuration);
+ MRJobType mrJobType = PhoenixConfigurationUtil.getMRJobType(configuration, MRJobType.QUERY.name());
+
+ String selectStatement;
+ switch (mrJobType) {
+ case UPDATE_STATS:
+ // This select statement indicates MR job for full table scan for stats collection
+ selectStatement = "SELECT * FROM " + PhoenixConfigurationUtil.getInputTableName(configuration);
+ break;
+ default:
+ selectStatement = PhoenixConfigurationUtil.getSelectStatement(configuration);
+ }
Preconditions.checkNotNull(selectStatement);
final PhoenixStatement pstmt = statement.unwrap(PhoenixStatement.class);
// Optimize the query plan so that we potentially use secondary indexes
final QueryPlan queryPlan = pstmt.optimizeQuery(selectStatement);
final Scan scan = queryPlan.getContext().getScan();
+
+ if (mrJobType == MRJobType.UPDATE_STATS) {
+ StatisticsUtil.setScanAttributes(scan, null);
+ }
+
// since we can't set a scn on connections with txn set TX_SCN attribute so that the max time range is set by BaseScannerRegionObserver
if (txnScnValue != null) {
scan.setAttribute(BaseScannerRegionObserver.TX_SCN, Bytes.toBytes(Long.valueOf(txnScnValue)));
@@ -194,9 +207,10 @@ public class PhoenixInputFormat<T extends DBWritable> extends InputFormat<NullWr
// setting the snapshot configuration
String snapshotName = configuration.get(PhoenixConfigurationUtil.SNAPSHOT_NAME_KEY);
- if (snapshotName != null)
- PhoenixConfigurationUtil.setSnapshotNameKey(queryPlan.getContext().getConnection().
- getQueryServices().getConfiguration(), snapshotName);
+ if (snapshotName != null) {
+ PhoenixConfigurationUtil.setSnapshotNameKey(queryPlan.getContext().getConnection().
+ getQueryServices().getConfiguration(), snapshotName);
+ }
// Initialize the query plan so it sets up the parallel scans
queryPlan.iterator(MapReduceParallelScanGrouper.getInstance());
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
index ac4ae67..180bdcb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
@@ -19,6 +19,7 @@ package org.apache.phoenix.mapreduce;
import java.io.IOException;
import java.sql.SQLException;
+import java.util.Arrays;
import java.util.List;
import org.apache.commons.logging.Log;
@@ -109,6 +110,7 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
final PhoenixInputSplit pSplit = (PhoenixInputSplit)split;
final List<Scan> scans = pSplit.getScans();
try {
+ LOG.info("Generating iterators for " + scans.size() + " scans in keyrange: " + pSplit.getKeyRange());
List<PeekingResultIterator> iterators = Lists.newArrayListWithExpectedSize(scans.size());
StatementContext ctx = queryPlan.getContext();
ReadMetricQueue readMetrics = ctx.getReadMetricsQueue();
@@ -134,6 +136,7 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
final TableSnapshotResultIterator tableSnapshotResultIterator = new TableSnapshotResultIterator(configuration, scan,
scanMetricsHolder);
peekingResultIterator = LookAheadResultIterator.wrap(tableSnapshotResultIterator);
+ LOG.info("Adding TableSnapshotResultIterator for scan: " + scan);
} else {
final TableResultIterator tableResultIterator =
new TableResultIterator(
@@ -141,8 +144,8 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
scanMetricsHolder, renewScannerLeaseThreshold, queryPlan,
MapReduceParallelScanGrouper.getInstance());
peekingResultIterator = LookAheadResultIterator.wrap(tableResultIterator);
+ LOG.info("Adding TableResultIterator for scan: " + scan);
}
-
iterators.add(peekingResultIterator);
}
ResultIterator iterator = queryPlan.useRoundRobinIterator() ? RoundRobinResultIterator.newIterator(iterators, queryPlan) : ConcatResultIterator.newIterator(iterators);
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
index 35f82c1..5a561ea 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java
@@ -153,9 +153,21 @@ public final class PhoenixConfigurationUtil {
public static final String MAPREDUCE_TENANT_ID = "phoenix.mapreduce.tenantid";
+ public static final String MAPREDUCE_JOB_TYPE = "phoenix.mapreduce.jobtype";
+
+ /**
+ * Determines type of Phoenix Map Reduce job.
+ * 1. QUERY allows running arbitrary queries without aggregates
+ * 2. UPDATE_STATS collects statistics for the table
+ */
+ public enum MRJobType {
+ QUERY,
+ UPDATE_STATS
+ }
+
public enum SchemaType {
TABLE,
- QUERY;
+ QUERY
}
private PhoenixConfigurationUtil(){
@@ -218,7 +230,12 @@ public final class PhoenixConfigurationUtil {
Preconditions.checkNotNull(configuration);
configuration.set(SCHEMA_TYPE, schemaType.name());
}
-
+
+ public static void setMRJobType(Configuration configuration, final MRJobType mrJobType) {
+ Preconditions.checkNotNull(configuration);
+ configuration.set(MAPREDUCE_JOB_TYPE, mrJobType.name());
+ }
+
public static void setPhysicalTableName(final Configuration configuration, final String tableName) {
Preconditions.checkNotNull(configuration);
Preconditions.checkNotNull(tableName);
@@ -286,7 +303,13 @@ public final class PhoenixConfigurationUtil {
Preconditions.checkNotNull(schemaTp);
return SchemaType.valueOf(schemaTp);
}
-
+
+ public static MRJobType getMRJobType(final Configuration configuration, String defaultMRJobType) {
+ final String mrJobType = configuration.get(MAPREDUCE_JOB_TYPE, defaultMRJobType);
+ Preconditions.checkNotNull(mrJobType);
+ return MRJobType.valueOf(mrJobType);
+ }
+
public static List<ColumnInfo> getUpsertColumnMetadataList(final Configuration configuration) throws SQLException {
Preconditions.checkNotNull(configuration);
List<ColumnInfo> columnMetadataList = null;
@@ -373,7 +396,7 @@ public final class PhoenixConfigurationUtil {
}
return selectColumnList;
}
-
+
public static String getSelectStatement(final Configuration configuration) throws SQLException {
Preconditions.checkNotNull(configuration);
String selectStmt = configuration.get(SELECT_STATEMENT);
@@ -389,7 +412,8 @@ public final class PhoenixConfigurationUtil {
configuration.set(SELECT_STATEMENT, selectStmt);
return selectStmt;
}
-
+
+
public static long getBatchSize(final Configuration configuration) throws SQLException {
Preconditions.checkNotNull(configuration);
long batchSize = configuration.getLong(UPSERT_BATCH_SIZE, DEFAULT_UPSERT_BATCH_SIZE);
@@ -619,6 +643,11 @@ public final class PhoenixConfigurationUtil {
configuration.setLong(SCRUTINY_EXECUTE_TIMESTAMP, ts);
}
+ public static void setSplitByStats(final Configuration configuration, boolean value) {
+ Preconditions.checkNotNull(configuration);
+ configuration.setBoolean(MAPREDUCE_SPLIT_BY_STATS, value);
+ }
+
public static String getDisableIndexes(Configuration configuration) {
Preconditions.checkNotNull(configuration);
return configuration.get(DISABLED_INDEXES);
@@ -674,4 +703,5 @@ public final class PhoenixConfigurationUtil {
Preconditions.checkNotNull(configuration);
configuration.set(MAPREDUCE_TENANT_ID, tenantId);
}
+
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
index 3462177..ecede80 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixMapReduceUtil.java
@@ -80,7 +80,7 @@ public final class PhoenixMapReduceUtil {
public static void setInput(final Job job, final Class<? extends DBWritable> inputClass, final String snapshotName, String tableName,
Path restoreDir, final String conditions, final String... fieldNames) throws
IOException {
- final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir);
+ final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir, SchemaType.QUERY);
if(conditions != null) {
PhoenixConfigurationUtil.setInputTableConditions(configuration, conditions);
}
@@ -99,13 +99,18 @@ public final class PhoenixMapReduceUtil {
public static void setInput(final Job job, final Class<? extends DBWritable> inputClass, final String snapshotName, String tableName,
Path restoreDir, String inputQuery) throws
IOException {
- final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir);
+ final Configuration configuration = setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir, SchemaType.QUERY);
if(inputQuery != null) {
PhoenixConfigurationUtil.setInputQuery(configuration, inputQuery);
}
}
+ public static void setInput(final Job job, final Class<? extends DBWritable> inputClass, final String snapshotName, String tableName,
+ Path restoreDir) {
+ setSnapshotInput(job, inputClass, snapshotName, tableName, restoreDir, SchemaType.QUERY);
+ }
+
/**
*
* @param job
@@ -115,7 +120,7 @@ public final class PhoenixMapReduceUtil {
* @param restoreDir a temporary dir to copy the snapshot files into
*/
private static Configuration setSnapshotInput(Job job, Class<? extends DBWritable> inputClass, String snapshotName,
- String tableName, Path restoreDir) {
+ String tableName, Path restoreDir, SchemaType schemaType) {
job.setInputFormatClass(PhoenixInputFormat.class);
final Configuration configuration = job.getConfiguration();
PhoenixConfigurationUtil.setInputClass(configuration, inputClass);
@@ -123,7 +128,7 @@ public final class PhoenixMapReduceUtil {
PhoenixConfigurationUtil.setInputTableName(configuration, tableName);
PhoenixConfigurationUtil.setRestoreDirKey(configuration, new Path(restoreDir, UUID.randomUUID().toString()).toString());
- PhoenixConfigurationUtil.setSchemaType(configuration, SchemaType.QUERY);
+ PhoenixConfigurationUtil.setSchemaType(configuration, schemaType);
return configuration;
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/GuidePostsCache.java b/phoenix-core/src/main/java/org/apache/phoenix/query/GuidePostsCache.java
index b78879b..abde79e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/GuidePostsCache.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/GuidePostsCache.java
@@ -17,7 +17,6 @@
package org.apache.phoenix.query;
import static org.apache.phoenix.query.QueryServices.STATS_COLLECTION_ENABLED;
-import static org.apache.phoenix.query.QueryServices.STATS_ENABLED_ATTRIB;
import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_STATS_COLLECTION_ENABLED;
import java.io.IOException;
@@ -70,8 +69,7 @@ public class GuidePostsCache {
final long maxTableStatsCacheSize = config.getLong(
QueryServices.STATS_MAX_CACHE_SIZE,
QueryServicesOptions.DEFAULT_STATS_MAX_CACHE_SIZE);
- final boolean isStatsEnabled = config.getBoolean(STATS_COLLECTION_ENABLED, DEFAULT_STATS_COLLECTION_ENABLED)
- && config.getBoolean(STATS_ENABLED_ATTRIB, true);
+ final boolean isStatsEnabled = config.getBoolean(STATS_COLLECTION_ENABLED, DEFAULT_STATS_COLLECTION_ENABLED);
cache = CacheBuilder.newBuilder()
// Expire entries a given amount of time after they were written
.expireAfterWrite(statsUpdateFrequency, TimeUnit.MILLISECONDS)
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index 8e17749..23cf292 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -192,9 +192,6 @@ public interface QueryServices extends SQLCloseable {
public static final String STATS_GUIDEPOST_PER_REGION_ATTRIB = "phoenix.stats.guidepost.per.region";
public static final String STATS_USE_CURRENT_TIME_ATTRIB = "phoenix.stats.useCurrentTime";
- @Deprecated // use STATS_COLLECTION_ENABLED config instead
- public static final String STATS_ENABLED_ATTRIB = "phoenix.stats.enabled";
-
public static final String RUN_UPDATE_STATS_ASYNC = "phoenix.update.stats.command.async";
public static final String STATS_SERVER_POOL_SIZE = "phoenix.stats.pool.size";
public static final String COMMIT_STATS_ASYNC = "phoenix.stats.commit.async";
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 74f8830..c60d31c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -218,6 +218,7 @@ import org.apache.phoenix.schema.PTable.QualifierEncodingScheme;
import org.apache.phoenix.schema.PTable.QualifierEncodingScheme.QualifierOutOfRangeException;
import org.apache.phoenix.schema.PTable.ViewType;
import org.apache.phoenix.schema.stats.GuidePostsKey;
+import org.apache.phoenix.schema.stats.StatisticsUtil;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PDate;
import org.apache.phoenix.schema.types.PInteger;
@@ -1317,21 +1318,9 @@ public class MetaDataClient {
TableRef tableRef = new TableRef(null, nonTxnLogicalTable, clientTimeStamp, false);
MutationPlan plan = compiler.compile(Collections.singletonList(tableRef), null, cfs, null, clientTimeStamp);
Scan scan = plan.getContext().getScan();
- scan.setCacheBlocks(false);
- scan.readAllVersions();
- scan.setAttribute(ANALYZE_TABLE, TRUE_BYTES);
+ StatisticsUtil.setScanAttributes(scan, statsProps);
boolean runUpdateStatsAsync = props.getBoolean(QueryServices.RUN_UPDATE_STATS_ASYNC, DEFAULT_RUN_UPDATE_STATS_ASYNC);
scan.setAttribute(RUN_UPDATE_STATS_ASYNC_ATTRIB, runUpdateStatsAsync ? TRUE_BYTES : FALSE_BYTES);
- if (statsProps != null) {
- Object gp_width = statsProps.get(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB);
- if (gp_width != null) {
- scan.setAttribute(BaseScannerRegionObserver.GUIDEPOST_WIDTH_BYTES, PLong.INSTANCE.toBytes(gp_width));
- }
- Object gp_per_region = statsProps.get(QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB);
- if (gp_per_region != null) {
- scan.setAttribute(BaseScannerRegionObserver.GUIDEPOST_PER_REGION, PInteger.INSTANCE.toBytes(gp_per_region));
- }
- }
MutationState mutationState = plan.execute();
rowCount = mutationState.getUpdateCount();
}
@@ -1722,6 +1711,7 @@ public class MetaDataClient {
}
PrimaryKeyConstraint pk = FACTORY.primaryKey(null, allPkColumns);
tableProps.put(MetaDataUtil.DATA_TABLE_NAME_PROP_NAME, dataTable.getName().getString());
+
CreateTableStatement tableStatement = FACTORY.createTable(indexTableName, statement.getProps(), columnDefs, pk, statement.getSplitNodes(), PTableType.INDEX, statement.ifNotExists(), null, null, statement.getBindCount(), null);
table = createTableInternal(tableStatement, splits, dataTable, null, null, MetaDataUtil.getViewIndexIdDataType(),null, null, allocateIndexId, statement.getIndexType(), asyncCreatedDate, tableProps, commonFamilyProps);
break;
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
index 7bc9dff..70379ee 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/DefaultStatisticsCollector.java
@@ -29,7 +29,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.client.Get;
@@ -50,7 +49,6 @@ import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
-import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.types.PInteger;
@@ -60,49 +58,39 @@ import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.SchemaUtil;
-import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
-import org.apache.phoenix.util.ServerUtil.ConnectionType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
/**
* A default implementation of the Statistics tracker that helps to collect stats like min key, max key and guideposts.
*/
-class DefaultStatisticsCollector implements StatisticsCollector {
- private static final Logger logger = LoggerFactory.getLogger(DefaultStatisticsCollector.class);
- private final Map<ImmutableBytesPtr, Pair<Long, GuidePostsInfoBuilder>> guidePostsInfoWriterMap = Maps.newHashMap();
+public class DefaultStatisticsCollector implements StatisticsCollector {
+
+ private static final Log LOG = LogFactory.getLog(DefaultStatisticsCollector.class);
+
+ final Map<ImmutableBytesPtr, Pair<Long, GuidePostsInfoBuilder>> guidePostsInfoWriterMap = Maps.newHashMap();
+ private final Table htable;
private StatisticsWriter statsWriter;
- private final Pair<Long, GuidePostsInfoBuilder> cachedGuidePosts;
- private final byte[] guidePostWidthBytes;
- private final byte[] guidePostPerRegionBytes;
+ final Pair<Long, GuidePostsInfoBuilder> cachedGuidePosts;
+ final byte[] guidePostWidthBytes;
+ final byte[] guidePostPerRegionBytes;
// Where to look for GUIDE_POSTS_WIDTH in SYSTEM.CATALOG
- private final byte[] ptableKey;
- private final RegionCoprocessorEnvironment env;
+ final byte[] ptableKey;
+
private long guidePostDepth;
private long maxTimeStamp = MetaDataProtocol.MIN_TABLE_TIMESTAMP;
- private static final Log LOG = LogFactory.getLog(DefaultStatisticsCollector.class);
private ImmutableBytesWritable currentRow;
- private final long clientTimeStamp;
private final String tableName;
private final boolean isViewIndexTable;
+ private final Region region;
+ private final Configuration configuration;
- DefaultStatisticsCollector(RegionCoprocessorEnvironment env, String tableName, long clientTimeStamp, byte[] family,
- byte[] gp_width_bytes, byte[] gp_per_region_bytes) throws IOException {
- this.env = env;
+ public DefaultStatisticsCollector(Configuration configuration, Region region, String tableName, byte[] family,
+ byte[] gp_width_bytes, byte[] gp_per_region_bytes, StatisticsWriter statsWriter, Table htable) {
+ this.configuration = configuration;
+ this.region = region;
this.guidePostWidthBytes = gp_width_bytes;
this.guidePostPerRegionBytes = gp_per_region_bytes;
- // Provides a means of clients controlling their timestamps to not use current time
- // when background tasks are updating stats. Instead we track the max timestamp of
- // the cells and use that.
- boolean useCurrentTime = env.getConfiguration().getBoolean(
- QueryServices.STATS_USE_CURRENT_TIME_ATTRIB,
- QueryServicesOptions.DEFAULT_STATS_USE_CURRENT_TIME);
- if (!useCurrentTime) {
- clientTimeStamp = DefaultStatisticsCollector.NO_TIMESTAMP;
- }
String pName = tableName;
// For view index, get GUIDE_POST_WIDTH from data physical table
// since there's no row representing those in SYSTEM.CATALOG.
@@ -113,7 +101,6 @@ class DefaultStatisticsCollector implements StatisticsCollector {
isViewIndexTable = false;
}
ptableKey = SchemaUtil.getTableKeyFromFullName(pName);
- this.clientTimeStamp = clientTimeStamp;
this.tableName = tableName;
// in a compaction we know the one family ahead of time
if (family != null) {
@@ -123,93 +110,126 @@ class DefaultStatisticsCollector implements StatisticsCollector {
} else {
cachedGuidePosts = null;
}
+
+ this.statsWriter = statsWriter;
+ this.htable = htable;
}
-
- private void initGuidepostDepth() throws IOException, ClassNotFoundException, SQLException {
- // First check is if guidepost info set on statement itself
+
+ @Override
+ public void init() throws IOException {
+ try {
+ initGuidepostDepth();
+ } catch (SQLException e) {
+ throw new IOException(e);
+ }
+ LOG.info("Initialization complete for " +
+ this.getClass() + " statistics collector for table " + tableName);
+ }
+
+ /**
+ * Determine the GPW for statistics collection for the table.
+ * The order of priority from highest to lowest is as follows
+ * 1. Value provided in UPDATE STATISTICS SQL statement (N/A for MR jobs)
+ * 2. GPW column in SYSTEM.CATALOG for the table is not null
+ * Inherits the value from base table for views and indexes (PHOENIX-4332)
+ * 3. Value from global configuration parameters from hbase-site.xml
+ *
+ * GPW of 0 disables the stats collection. If stats were previously collected, this task
+ * would attempt to delete entries from SYSTEM.STATS table. Not reading '0' from SYSTEM.CATALOG
+ * would mean the fall back to global value which is defaulted to DEFAULT_STATS_GUIDEPOST_PER_REGION
+ */
+ private void initGuidepostDepth() throws IOException, SQLException {
if (guidePostPerRegionBytes != null || guidePostWidthBytes != null) {
- int guidepostPerRegion = 0;
- long guidepostWidth = QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES;
- if (guidePostPerRegionBytes != null) {
- guidepostPerRegion = PInteger.INSTANCE.getCodec().decodeInt(guidePostPerRegionBytes, 0, SortOrder.getDefault());
- }
- if (guidePostWidthBytes != null) {
- guidepostWidth = PLong.INSTANCE.getCodec().decodeInt(guidePostWidthBytes, 0, SortOrder.getDefault());
- }
- this.guidePostDepth = StatisticsUtil.getGuidePostDepth(guidepostPerRegion, guidepostWidth,
- env.getRegion().getTableDescriptor());
+ getGuidePostDepthFromStatement();
+ LOG.info("Guide post depth determined from SQL statement: " + guidePostDepth);
} else {
+ long guidepostWidth = getGuidePostDepthFromSystemCatalog();
+ if (guidepostWidth >= 0) {
+ this.guidePostDepth = guidepostWidth;
+ LOG.info("Guide post depth determined from SYSTEM.CATALOG: " + guidePostDepth);
+ } else {
+ this.guidePostDepth = StatisticsUtil.getGuidePostDepth(
+ configuration.getInt(
+ QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB,
+ QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_PER_REGION),
+ configuration.getLong(
+ QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB,
+ QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES),
+ region.getTableDescriptor());
+ LOG.info("Guide post depth determined from global configuration: " + guidePostDepth);
+ }
+ }
+
+ }
+
+ private long getGuidePostDepthFromSystemCatalog() throws IOException, SQLException {
+ try {
long guidepostWidth = -1;
- Table htable = null;
- try {
- // Next check for GUIDE_POST_WIDTH on table
- htable = ConnectionFactory.getConnection(ConnectionType.DEFAULT_SERVER_CONNECTION, env).getTable(
- SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, env.getConfiguration()));
- Get get = new Get(ptableKey);
- get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH_BYTES);
- Result result = htable.get(get);
- if (!result.isEmpty()) {
- Cell cell = result.listCells().get(0);
- guidepostWidth = PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(), cell.getValueOffset(), SortOrder.getDefault());
- } else if (!isViewIndexTable) {
- /*
- * The table we are collecting stats for is potentially a base table, or local
- * index or a global index. For view indexes, we rely on the the guide post
- * width column in the parent data table's metadata which we already tried
- * retrieving above.
- */
- try (Connection conn =
- QueryUtil.getConnectionOnServer(env.getConfiguration())) {
- PTable table = PhoenixRuntime.getTable(conn, tableName);
- if (table.getType() == PTableType.INDEX
- && table.getIndexType() == IndexType.GLOBAL) {
- /*
- * For global indexes, we need to get the parentName first and then
- * fetch guide post width configured for the parent table.
- */
- PName parentName = table.getParentName();
- byte[] parentKey =
- SchemaUtil.getTableKeyFromFullName(parentName.getString());
- get = new Get(parentKey);
- get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
+ Get get = new Get(ptableKey);
+ get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES, PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH_BYTES);
+ Result result = htable.get(get);
+ if (!result.isEmpty()) {
+ Cell cell = result.listCells().get(0);
+ guidepostWidth = PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(), cell.getValueOffset(), SortOrder.getDefault());
+ } else if (!isViewIndexTable) {
+ /*
+ * The table we are collecting stats for is potentially a base table, or local
+ * index or a global index. For view indexes, we rely on the the guide post
+ * width column in the parent data table's metadata which we already tried
+ * retrieving above.
+ */
+ try (Connection conn =
+ QueryUtil.getConnectionOnServer(configuration)) {
+ PTable table = PhoenixRuntime.getTable(conn, tableName);
+ if (table.getType() == PTableType.INDEX
+ && table.getIndexType() == PTable.IndexType.GLOBAL) {
+ /*
+ * For global indexes, we need to get the parentName first and then
+ * fetch guide post width configured for the parent table.
+ */
+ PName parentName = table.getParentName();
+ byte[] parentKey =
+ SchemaUtil.getTableKeyFromFullName(parentName.getString());
+ get = new Get(parentKey);
+ get.addColumn(PhoenixDatabaseMetaData.TABLE_FAMILY_BYTES,
PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH_BYTES);
- result = htable.get(get);
- if (!result.isEmpty()) {
- Cell cell = result.listCells().get(0);
- guidepostWidth =
- PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(),
+ result = htable.get(get);
+ if (!result.isEmpty()) {
+ Cell cell = result.listCells().get(0);
+ guidepostWidth =
+ PLong.INSTANCE.getCodec().decodeLong(cell.getValueArray(),
cell.getValueOffset(), SortOrder.getDefault());
- }
}
}
-
- }
- } finally {
- if (htable != null) {
- try {
- htable.close();
- } catch (IOException e) {
- LOG.warn("Failed to close " + htable.getName(), e);
- }
+ } catch (ClassNotFoundException e) {
+ throw new IOException(e);
}
}
- if (guidepostWidth >= 0) {
- this.guidePostDepth = guidepostWidth;
- } else {
- // Last use global config value
- Configuration config = env.getConfiguration();
- this.guidePostDepth = StatisticsUtil.getGuidePostDepth(
- config.getInt(
- QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB,
- QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_PER_REGION),
- config.getLong(
- QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB,
- QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES),
- env.getRegion().getTableDescriptor());
+ return guidepostWidth;
+ } finally {
+ if (htable != null) {
+ try {
+ htable.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close " + htable.getName(), e);
+ }
}
}
}
+ private void getGuidePostDepthFromStatement() {
+ int guidepostPerRegion = 0;
+ long guidepostWidth = QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES;
+ if (guidePostPerRegionBytes != null) {
+ guidepostPerRegion = PInteger.INSTANCE.getCodec().decodeInt(guidePostPerRegionBytes, 0, SortOrder.getDefault());
+ }
+ if (guidePostWidthBytes != null) {
+ guidepostWidth = PLong.INSTANCE.getCodec().decodeInt(guidePostWidthBytes, 0, SortOrder.getDefault());
+ }
+ this.guidePostDepth = StatisticsUtil.getGuidePostDepth(guidepostPerRegion, guidepostWidth,
+ region.getTableDescriptor());
+ }
+
@Override
public long getMaxTimeStamp() {
return maxTimeStamp;
@@ -223,68 +243,62 @@ class DefaultStatisticsCollector implements StatisticsCollector {
}
@Override
- public void updateStatistic(Region region, Scan scan) {
+ public void updateStatistics(Region region, Scan scan) {
try {
- ArrayList<Mutation> mutations = new ArrayList<Mutation>();
- writeStatistics(region, true, mutations, EnvironmentEdgeManager.currentTimeMillis(), scan);
- if (logger.isDebugEnabled()) {
- logger.debug("Committing new stats for the region " + region.getRegionInfo());
- }
+ List<Mutation> mutations = new ArrayList<Mutation>();
+ writeStatistics(region, true, mutations,
+ EnvironmentEdgeManager.currentTimeMillis(), scan);
commitStats(mutations);
} catch (IOException e) {
- logger.error("Unable to commit new stats", e);
+ LOG.error("Unable to update SYSTEM.STATS table.", e);
}
}
private void writeStatistics(final Region region, boolean delete, List<Mutation> mutations, long currentTime, Scan scan)
throws IOException {
- try {
- Set<ImmutableBytesPtr> fams = guidePostsInfoWriterMap.keySet();
- // Update the statistics table.
- // Delete statistics for a region if no guide posts are collected for that region during
- // UPDATE STATISTICS. This will not impact a stats collection of single column family during
- // compaction as guidePostsInfoWriterMap cannot be empty in this case.
- if (cachedGuidePosts == null) {
- // We're either collecting stats for the data table or the local index table, but not both
- // We can determine this based on the column families in the scan being prefixed with the
- // local index column family prefix. We always explicitly specify the local index column
- // families when we're collecting stats for a local index.
- boolean collectingForLocalIndex = scan != null && !scan.getFamilyMap().isEmpty() && MetaDataUtil.isLocalIndexFamily(scan.getFamilyMap().keySet().iterator().next());
- for (Store store : region.getStores()) {
- ImmutableBytesPtr cfKey = new ImmutableBytesPtr(store.getColumnFamilyDescriptor().getName());
- boolean isLocalIndexStore = MetaDataUtil.isLocalIndexFamily(cfKey);
- if (isLocalIndexStore != collectingForLocalIndex) {
- continue;
- }
- if (!guidePostsInfoWriterMap.containsKey(cfKey)) {
- Pair<Long, GuidePostsInfoBuilder> emptyGps = new Pair<Long, GuidePostsInfoBuilder>(0l, new GuidePostsInfoBuilder());
- guidePostsInfoWriterMap.put(cfKey, emptyGps);
- }
+ Set<ImmutableBytesPtr> fams = guidePostsInfoWriterMap.keySet();
+ // Update the statistics table.
+ // Delete statistics for a region if no guide posts are collected for that region during
+ // UPDATE STATISTICS. This will not impact a stats collection of single column family during
+ // compaction as guidePostsInfoWriterMap cannot be empty in this case.
+ if (cachedGuidePosts == null) {
+ // We're either collecting stats for the data table or the local index table, but not both
+ // We can determine this based on the column families in the scan being prefixed with the
+ // local index column family prefix. We always explicitly specify the local index column
+ // families when we're collecting stats for a local index.
+ boolean collectingForLocalIndex = scan != null &&
+ !scan.getFamilyMap().isEmpty() &&
+ MetaDataUtil.isLocalIndexFamily(scan.getFamilyMap().keySet().iterator().next());
+ for (Store store : region.getStores()) {
+ ImmutableBytesPtr cfKey = new ImmutableBytesPtr(store.getColumnFamilyDescriptor().getName());
+ boolean isLocalIndexStore = MetaDataUtil.isLocalIndexFamily(cfKey);
+ if (isLocalIndexStore != collectingForLocalIndex) {
+ continue;
}
- }
- for (ImmutableBytesPtr fam : fams) {
- if (delete) {
- if (logger.isDebugEnabled()) {
- logger.debug("Deleting the stats for the region " + region.getRegionInfo());
- }
- statsWriter.deleteStatsForRegion(region, this, fam, mutations);
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Adding new stats for the region " + region.getRegionInfo());
- }
- // If we've disabled stats, don't write any, just delete them
- if (this.guidePostDepth > 0) {
- statsWriter.addStats(this, fam, mutations);
+ if (!guidePostsInfoWriterMap.containsKey(cfKey)) {
+ Pair<Long, GuidePostsInfoBuilder> emptyGps = new Pair<Long, GuidePostsInfoBuilder>(0l, new GuidePostsInfoBuilder());
+ guidePostsInfoWriterMap.put(cfKey, emptyGps);
}
}
- } catch (IOException e) {
- logger.error("Failed to update statistics table!", e);
- throw e;
+ }
+ for (ImmutableBytesPtr fam : fams) {
+ if (delete) {
+ statsWriter.deleteStatsForRegion(region, this, fam, mutations);
+ LOG.info("Generated " + mutations.size() + " mutations to delete existing stats");
+ }
+
+ // If we've disabled stats, don't write any, just delete them
+ if (this.guidePostDepth > 0) {
+ int oldSize = mutations.size();
+ statsWriter.addStats(this, fam, mutations, guidePostDepth);
+ LOG.info("Generated " + (mutations.size() - oldSize) + " mutations for new stats");
+ }
}
}
private void commitStats(List<Mutation> mutations) throws IOException {
statsWriter.commitStats(mutations, this);
+ LOG.info("Committed " + mutations.size() + " mutations for stats");
}
/**
@@ -349,31 +363,6 @@ class DefaultStatisticsCollector implements StatisticsCollector {
}
@Override
- public InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env, Store store,
- InternalScanner s) throws IOException {
- // See if this is for Major compaction
- if (logger.isDebugEnabled()) {
- logger.debug("Compaction scanner created for stats");
- }
- ImmutableBytesPtr cfKey = new ImmutableBytesPtr(store.getColumnFamilyDescriptor().getName());
- // Potentially perform a cross region server get in order to use the correct guide posts
- // width for the table being compacted.
- init();
- StatisticsScanner scanner = new StatisticsScanner(this, statsWriter, env, s, cfKey);
- return scanner;
- }
-
- @Override
- public void init() throws IOException {
- try {
- initGuidepostDepth();
- } catch (ClassNotFoundException | SQLException e) {
- throw new IOException("Unable to initialize the guide post depth", e);
- }
- this.statsWriter = StatisticsWriter.newWriter(env, tableName, clientTimeStamp, guidePostDepth);
- }
-
- @Override
public GuidePostsInfo getGuidePosts(ImmutableBytesPtr fam) {
Pair<Long, GuidePostsInfoBuilder> pair = guidePostsInfoWriterMap.get(fam);
if (pair != null) {
@@ -382,9 +371,25 @@ class DefaultStatisticsCollector implements StatisticsCollector {
return null;
}
- @VisibleForTesting // Don't call this method anywhere else
+ @Override
public long getGuidePostDepth() {
return guidePostDepth;
}
+ @Override
+ public StatisticsWriter getStatisticsWriter() {
+ return statsWriter;
+ }
+
+ @Override
+ public InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env,
+ Store store, InternalScanner delegate) {
+
+ ImmutableBytesPtr cfKey =
+ new ImmutableBytesPtr(store.getColumnFamilyDescriptor().getName());
+ LOG.info("StatisticsScanner created for table: "
+ + tableName + " CF: " + store.getColumnFamilyName());
+ return new StatisticsScanner(this, statsWriter, env, delegate, cfKey);
+ }
+
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
index 74d1710..4a8d25f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/NoOpStatisticsCollector.java
@@ -45,7 +45,7 @@ public class NoOpStatisticsCollector implements StatisticsCollector {
}
@Override
- public void updateStatistic(Region region, Scan scan) {
+ public void updateStatistics(Region region, Scan scan) {
// No-op
}
@@ -55,12 +55,6 @@ public class NoOpStatisticsCollector implements StatisticsCollector {
}
@Override
- public InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env, Store store,
- InternalScanner delegate) throws IOException {
- return delegate;
- }
-
- @Override
public void init() {
// No-op
}
@@ -68,4 +62,20 @@ public class NoOpStatisticsCollector implements StatisticsCollector {
@Override public GuidePostsInfo getGuidePosts(ImmutableBytesPtr fam) {
return null;
}
+
+ @Override
+ public long getGuidePostDepth() {
+ return -1;
+ }
+
+ @Override
+ public StatisticsWriter getStatisticsWriter() {
+ return null;
+ }
+
+ @Override
+ public InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env,
+ Store store, InternalScanner delegate) {
+ return delegate;
+ }
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
index 60e83a8..0fd6390 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollector.java
@@ -45,7 +45,7 @@ public interface StatisticsCollector extends Closeable {
/**
* Write the collected statistics for the given region over the scan provided.
*/
- void updateStatistic(Region region, Scan scan);
+ void updateStatistics(Region region, Scan scan);
/**
* Collect statistics for the given list of cells. This method can be called multiple times
@@ -55,12 +55,6 @@ public interface StatisticsCollector extends Closeable {
void collectStatistics(List<Cell> results);
/**
- * Wrap a compaction scanner with a scanner that will collect statistics using this instance.
- */
- InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env, Store store,
- InternalScanner delegate) throws IOException;
-
- /**
* Called before beginning the collection of statistics through {@link #collectStatistics(List)}
* @throws IOException
*/
@@ -70,4 +64,21 @@ public interface StatisticsCollector extends Closeable {
* Retrieve the calculated guide post info for the given column family.
*/
GuidePostsInfo getGuidePosts(ImmutableBytesPtr fam);
+
+ /**
+ * Retrieve the guide post depth during stats collection
+ */
+ long getGuidePostDepth();
+
+ /**
+ * Retrieve the object that manages statistics persistence
+ */
+ StatisticsWriter getStatisticsWriter();
+
+ /**
+ * Wrap a compaction scanner with a scanner that will collect statistics using this instance.
+ */
+ InternalScanner createCompactionScanner(RegionCoprocessorEnvironment env, Store store,
+ InternalScanner delegate);
+
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
index 4e37e5c..210ea79 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsCollectorFactory.java
@@ -18,12 +18,16 @@
package org.apache.phoenix.schema.stats;
import static org.apache.phoenix.query.QueryServices.STATS_COLLECTION_ENABLED;
-import static org.apache.phoenix.query.QueryServices.STATS_ENABLED_ATTRIB;
import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_STATS_COLLECTION_ENABLED;
import java.io.IOException;
+import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
/**
* Provides new {@link StatisticsCollector} instances based on configuration settings for a
@@ -48,8 +52,11 @@ public class StatisticsCollectorFactory {
byte[] storeName, byte[] guidepostWidthBytes,
byte[] guidepostsPerRegionBytes) throws IOException {
if (statisticsEnabled(env)) {
- return new DefaultStatisticsCollector(env, tableName, clientTimeStamp, storeName,
- guidepostWidthBytes, guidepostsPerRegionBytes);
+ StatisticsWriter statsWriter = StatisticsWriter.newWriter(env, tableName, clientTimeStamp);
+ Table table = ConnectionFactory.getConnection(ConnectionType.DEFAULT_SERVER_CONNECTION, env).getTable(
+ SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES, env.getConfiguration()));
+ return new DefaultStatisticsCollector(env.getConfiguration(), env.getRegion(), tableName,
+ storeName,guidepostWidthBytes, guidepostsPerRegionBytes, statsWriter, table);
} else {
return new NoOpStatisticsCollector();
}
@@ -62,11 +69,7 @@ public class StatisticsCollectorFactory {
* use case for that.
*/
private static boolean statisticsEnabled(RegionCoprocessorEnvironment env) {
- return (env.getConfiguration().getBoolean(STATS_COLLECTION_ENABLED,
- DEFAULT_STATS_COLLECTION_ENABLED)
- // old config left here for backward compatibility. TODO: remove in the next major release
- && env.getConfiguration().getBoolean(STATS_ENABLED_ATTRIB, true)
- )
+ return (env.getConfiguration().getBoolean(STATS_COLLECTION_ENABLED, DEFAULT_STATS_COLLECTION_ENABLED))
&& StatisticsUtil.isStatsEnabled(env.getRegionInfo().getTable());
}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsScanner.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsScanner.java
index 3f5809b..36eb9bd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsScanner.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsScanner.java
@@ -158,7 +158,8 @@ public class StatisticsScanner implements InternalScanner {
LOG.debug("Adding new stats for the region " + regionInfo.getRegionNameAsString()
+ " as part of major compaction");
}
- getStatisticsWriter().addStats(tracker, family, mutations);
+ getStatisticsWriter().addStats(tracker, family,
+ mutations, tracker.getGuidePostDepth());
if (LOG.isDebugEnabled()) {
LOG.debug("Committing new stats for the region " + regionInfo.getRegionNameAsString()
+ " as part of major compaction");
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsUtil.java
index 3e8ad7d..7fd1fcf 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsUtil.java
@@ -16,9 +16,12 @@
* limitations under the License.
*/
package org.apache.phoenix.schema.stats;
+import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.ANALYZE_TABLE;
+import static org.apache.phoenix.schema.types.PDataType.TRUE_BYTES;
import static org.apache.phoenix.util.SchemaUtil.getVarCharLength;
import java.io.IOException;
+import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hbase.Cell;
@@ -32,10 +35,13 @@ import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.types.PInteger;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.MetaDataUtil;
@@ -226,6 +232,21 @@ public class StatisticsUtil {
public static boolean isStatsEnabled(TableName tableName) {
return !DISABLE_STATS.contains(tableName);
}
-
+
+ public static void setScanAttributes(Scan scan, Map<String, Object> statsProps) {
+ scan.setCacheBlocks(false);
+ scan.readAllVersions();
+ scan.setAttribute(ANALYZE_TABLE, TRUE_BYTES);
+ if (statsProps != null) {
+ Object gp_width = statsProps.get(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB);
+ if (gp_width != null) {
+ scan.setAttribute(BaseScannerRegionObserver.GUIDEPOST_WIDTH_BYTES, PLong.INSTANCE.toBytes(gp_width));
+ }
+ Object gp_per_region = statsProps.get(QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB);
+ if (gp_per_region != null) {
+ scan.setAttribute(BaseScannerRegionObserver.GUIDEPOST_PER_REGION, PInteger.INSTANCE.toBytes(gp_per_region));
+ }
+ }
+ }
}
\ No newline at end of file
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java
index 6e0ae81..bb2c22f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java
@@ -27,11 +27,14 @@ import java.io.EOFException;
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
import java.sql.Date;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
@@ -51,8 +54,11 @@ import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.types.PDate;
import org.apache.phoenix.schema.types.PLong;
import org.apache.phoenix.util.ByteUtil;
@@ -61,15 +67,28 @@ import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PrefixByteDecoder;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.ServerUtil;
-import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
-import org.apache.phoenix.util.ServerUtil.ConnectionType;
import com.google.protobuf.ServiceException;
+import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
/**
* Wrapper to access the statistics table SYSTEM.STATS using the HTable.
*/
public class StatisticsWriter implements Closeable {
+
+ public static StatisticsWriter newWriter(PhoenixConnection conn, String tableName, long clientTimeStamp)
+ throws SQLException {
+ Configuration configuration = conn.getQueryServices().getConfiguration();
+ long newClientTimeStamp = determineClientTimeStamp(configuration, clientTimeStamp);
+ TableName physicalTableName = SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES, configuration);
+ Table statsWriterTable = conn.getQueryServices().getTable(physicalTableName.getName());
+ Table statsReaderTable = conn.getQueryServices().getTable(physicalTableName.getName());
+ StatisticsWriter statsTable = new StatisticsWriter(statsReaderTable, statsWriterTable, tableName,
+ newClientTimeStamp);
+ return statsTable;
+ }
+
/**
* @param tableName
* TODO
@@ -79,19 +98,34 @@ public class StatisticsWriter implements Closeable {
* @throws IOException
* if the table cannot be created due to an underlying HTable creation error
*/
- public static StatisticsWriter newWriter(RegionCoprocessorEnvironment env, String tableName, long clientTimeStamp, long guidePostDepth)
+ public static StatisticsWriter newWriter(RegionCoprocessorEnvironment env, String tableName, long clientTimeStamp)
throws IOException {
- if (clientTimeStamp == HConstants.LATEST_TIMESTAMP) {
- clientTimeStamp = EnvironmentEdgeManager.currentTimeMillis();
- }
+ Configuration configuration = env.getConfiguration();
+ long newClientTimeStamp = determineClientTimeStamp(configuration, clientTimeStamp);
Table statsWriterTable = ConnectionFactory.getConnection(ConnectionType.DEFAULT_SERVER_CONNECTION, env).getTable(
SchemaUtil.getPhysicalTableName(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES, env.getConfiguration()));
Table statsReaderTable = ServerUtil.getHTableForCoprocessorScan(env, statsWriterTable);
StatisticsWriter statsTable = new StatisticsWriter(statsReaderTable, statsWriterTable, tableName,
- clientTimeStamp, guidePostDepth);
+ newClientTimeStamp);
return statsTable;
}
+ // Provides a means of clients controlling their timestamps to not use current time
+ // when background tasks are updating stats. Instead we track the max timestamp of
+ // the cells and use that.
+ private static long determineClientTimeStamp(Configuration configuration, long clientTimeStamp) {
+ boolean useCurrentTime = configuration.getBoolean(
+ QueryServices.STATS_USE_CURRENT_TIME_ATTRIB,
+ QueryServicesOptions.DEFAULT_STATS_USE_CURRENT_TIME);
+ if (!useCurrentTime) {
+ clientTimeStamp = DefaultStatisticsCollector.NO_TIMESTAMP;
+ }
+ if (clientTimeStamp == HConstants.LATEST_TIMESTAMP) {
+ clientTimeStamp = EnvironmentEdgeManager.currentTimeMillis();
+ }
+ return clientTimeStamp;
+ }
+
private final Table statsWriterTable;
// In HBase 0.98.4 or above, the reader and writer will be the same.
// In pre HBase 0.98.4, there was a bug in using the HTable returned
@@ -99,15 +133,13 @@ public class StatisticsWriter implements Closeable {
private final Table statsReaderTable;
private final byte[] tableName;
private final long clientTimeStamp;
- private final long guidePostDepth;
-
- private StatisticsWriter(Table statsReaderTable, Table statsWriterTable, String tableName,
- long clientTimeStamp, long guidePostDepth) {
+
+ private StatisticsWriter(Table statsReaderTable,
+ Table statsWriterTable, String tableName, long clientTimeStamp) {
this.statsReaderTable = statsReaderTable;
this.statsWriterTable = statsWriterTable;
this.tableName = Bytes.toBytes(tableName);
this.clientTimeStamp = clientTimeStamp;
- this.guidePostDepth = guidePostDepth;
}
/**
@@ -135,8 +167,8 @@ public class StatisticsWriter implements Closeable {
* remaining list of stats to update
*/
@SuppressWarnings("deprecation")
- public void addStats(StatisticsCollector tracker, ImmutableBytesPtr cfKey, List<Mutation> mutations)
- throws IOException {
+ public void addStats(StatisticsCollector tracker, ImmutableBytesPtr cfKey,
+ List<Mutation> mutations, long guidePostDepth) throws IOException {
if (tracker == null) { return; }
boolean useMaxTimeStamp = clientTimeStamp == DefaultStatisticsCollector.NO_TIMESTAMP;
long timeStamp = clientTimeStamp;
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatsCollectionDisabledOnServerException.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatsCollectionDisabledOnServerException.java
new file mode 100644
index 0000000..05d66f6
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatsCollectionDisabledOnServerException.java
@@ -0,0 +1,33 @@
+/*
+ * 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.schema.stats;
+
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+
+import java.sql.SQLException;
+
+public class StatsCollectionDisabledOnServerException extends SQLException {
+ private static final long serialVersionUID = 1L;
+ private static SQLExceptionCode code = SQLExceptionCode.STATS_COLLECTION_DISABLED_ON_SERVER;
+
+ public StatsCollectionDisabledOnServerException() {
+ super(new SQLExceptionInfo.Builder(code)
+ .setMessage(code.getMessage()).build().toString(),
+ code.getSQLState(), code.getErrorCode());
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java
new file mode 100644
index 0000000..f84b859
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/UpdateStatisticsTool.java
@@ -0,0 +1,223 @@
+/*
+ * 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.schema.stats;
+
+import org.antlr.runtime.CharStream;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
+import org.apache.hadoop.hbase.metrics.Gauge;
+import org.apache.hadoop.hbase.metrics.impl.MetricRegistriesImpl;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
+import org.apache.hadoop.mapreduce.lib.db.DBInputFormat.NullDBWritable;
+import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.htrace.SpanReceiver;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MRJobType;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.SchemaType;
+import org.apache.phoenix.mapreduce.util.PhoenixMapReduceUtil;
+import org.apache.tephra.TransactionNotInProgressException;
+import org.apache.tephra.TransactionSystemClient;
+import org.apache.tephra.hbase.coprocessor.TransactionProcessor;
+import org.apache.thrift.transport.TTransportException;
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.discovery.DiscoveryServiceClient;
+import org.apache.twill.discovery.ZKDiscoveryService;
+import org.apache.twill.zookeeper.ZKClient;
+import org.joda.time.Chronology;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.UUID;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
+
+/**
+ * Tool to collect table level statistics on HBase snapshot
+ */
+public class UpdateStatisticsTool extends Configured implements Tool {
+
+ private static final Logger LOG = LoggerFactory.getLogger(UpdateStatisticsTool.class);
+
+ private static final Option TABLE_NAME_OPTION = new Option("t", "table", true,
+ "Phoenix Table Name");
+ private static final Option SNAPSHOT_NAME_OPTION = new Option("s", "snapshot", true,
+ "HBase Snapshot Name");
+ private static final Option RESTORE_DIR_OPTION = new Option("d", "restore-dir", true,
+ "Restore Directory for HBase snapshot");
+ private static final Option RUN_FOREGROUND_OPTION =
+ new Option("runfg", "run-foreground", false,
+ "If specified, runs UpdateStatisticsTool in Foreground. Default - Runs the build in background.");
+ private static final Option HELP_OPTION = new Option("h", "help", false, "Help");
+
+ private Configuration conf;
+ private String tableName;
+ private String snapshotName;
+ private Path restoreDir;
+ private boolean isForeground;
+
+ @Override
+ public int run(String[] args) throws Exception {
+ parseArgs(args);
+ Job job = configureJob(conf, tableName, snapshotName, restoreDir);
+ TableMapReduceUtil.initCredentials(job);
+ return runJob(job, isForeground);
+ }
+
+ private void parseArgs(String[] args) {
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parseOptions(args);
+ } catch (IllegalStateException e) {
+ printHelpAndExit(e.getMessage(), getOptions());
+ }
+
+ conf = HBaseConfiguration.create();
+ tableName = cmdLine.getOptionValue(TABLE_NAME_OPTION.getOpt());
+ snapshotName = cmdLine.getOptionValue(SNAPSHOT_NAME_OPTION.getOpt());
+ String restoreDirOptionValue = cmdLine.getOptionValue(RESTORE_DIR_OPTION.getOpt());
+ if (restoreDirOptionValue == null) {
+ restoreDirOptionValue = conf.get(FS_DEFAULT_NAME_KEY) + "/tmp";
+ }
+ restoreDir = new Path(restoreDirOptionValue);
+ isForeground = cmdLine.hasOption(RUN_FOREGROUND_OPTION.getOpt());
+ }
+
+ Job configureJob(Configuration conf, String tableName,
+ String snapshotName, Path restoreDir) throws Exception {
+ Job job = Job.getInstance(conf, "Update statistics for " + tableName);
+ PhoenixMapReduceUtil.setInput(job, NullDBWritable.class,
+ snapshotName, tableName, restoreDir);
+
+ PhoenixConfigurationUtil.setMRJobType(job.getConfiguration(), MRJobType.UPDATE_STATS);
+ // DO NOT allow mapper splits using statistics since it may result into many smaller chunks
+ PhoenixConfigurationUtil.setSplitByStats(job.getConfiguration(), false);
+
+ job.setJarByClass(UpdateStatisticsTool.class);
+ job.setMapperClass(TableSnapshotMapper.class);
+ job.setMapOutputKeyClass(NullWritable.class);
+ job.setMapOutputValueClass(NullWritable.class);
+ job.setOutputFormatClass(NullOutputFormat.class);
+ job.setNumReduceTasks(0);
+ TableMapReduceUtil.addDependencyJars(job);
+ TableMapReduceUtil.addDependencyJarsForClasses(job.getConfiguration(), PhoenixConnection.class, Chronology.class,
+ CharStream.class, TransactionSystemClient.class, TransactionNotInProgressException.class,
+ ZKClient.class, DiscoveryServiceClient.class, ZKDiscoveryService.class,
+ Cancellable.class, TTransportException.class, SpanReceiver.class, TransactionProcessor.class, Gauge.class, MetricRegistriesImpl.class);
+ LOG.info("UpdateStatisticsTool running for: " + tableName
+ + " on snapshot: " + snapshotName + " with restore dir: " + restoreDir);
+
+ return job;
+ }
+
+ int runJob(Job job, boolean isForeground) throws Exception {
+ if (isForeground) {
+ LOG.info("Running UpdateStatisticsTool in Foreground. " +
+ "Runs full table scans. This may take a long time!.");
+ return (job.waitForCompletion(true)) ? 0 : 1;
+ } else {
+ LOG.info("Running UpdateStatisticsTool in Background - Submit async and exit");
+ job.submit();
+ return 0;
+ }
+ }
+
+ private void printHelpAndExit(String errorMessage, Options options) {
+ System.err.println(errorMessage);
+ printHelpAndExit(options, 1);
+ }
+
+ private void printHelpAndExit(Options options, int exitCode) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("help", options);
+ System.exit(exitCode);
+ }
+
+ /**
+ * Parses the commandline arguments, throws IllegalStateException if mandatory arguments are
+ * missing.
+ * @param args supplied command line arguments
+ * @return the parsed command line
+ */
+ private CommandLine parseOptions(String[] args) {
+
+ final Options options = getOptions();
+
+ CommandLineParser parser = new PosixParser();
+ CommandLine cmdLine = null;
+ try {
+ cmdLine = parser.parse(options, args);
+ } catch (ParseException e) {
+ printHelpAndExit("Error parsing command line options: " + e.getMessage(), options);
+ }
+
+ if (cmdLine.hasOption(HELP_OPTION.getOpt())) {
+ printHelpAndExit(options, 0);
+ }
+
+ if (!cmdLine.hasOption(TABLE_NAME_OPTION.getOpt())) {
+ throw new IllegalStateException(TABLE_NAME_OPTION.getLongOpt() + " is a mandatory "
+ + "parameter");
+ }
+
+ if (!cmdLine.hasOption(SNAPSHOT_NAME_OPTION.getOpt())) {
+ throw new IllegalStateException(SNAPSHOT_NAME_OPTION.getLongOpt() + " is a mandatory "
+ + "parameter");
+ }
+
+ return cmdLine;
+ }
+
+ private Options getOptions() {
+ final Options options = new Options();
+ options.addOption(TABLE_NAME_OPTION);
+ options.addOption(SNAPSHOT_NAME_OPTION);
+ options.addOption(HELP_OPTION);
+ options.addOption(RESTORE_DIR_OPTION);
+ options.addOption(RUN_FOREGROUND_OPTION);
+ return options;
+ }
+
+ public static class TableSnapshotMapper
+ extends Mapper<NullWritable, NullDBWritable, NullWritable, NullWritable> {
+
+ @Override
+ protected void map(NullWritable key, NullDBWritable value,
+ Context context) {
+ }
+ }
+
+ public static void main(String[] args) throws Exception {
+ ToolRunner.run(new UpdateStatisticsTool(), args);
+ }
+}
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 334b65c..4d2d44b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -211,7 +211,8 @@ public class ServerUtil {
}
}
- public static Table getHTableForCoprocessorScan (RegionCoprocessorEnvironment env, Table writerTable) throws IOException {
+ public static Table getHTableForCoprocessorScan (RegionCoprocessorEnvironment env,
+ Table writerTable) throws IOException {
if (coprocessorScanWorks(env)) {
return writerTable;
}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtilTest.java b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtilTest.java
index 0c4c004..cb5efa1 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtilTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtilTest.java
@@ -25,6 +25,7 @@ import java.sql.DriverManager;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.mapreduce.Job;
+import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.MRJobType;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil.SchemaType;
import org.apache.phoenix.query.BaseConnectionlessQueryTest;
import org.apache.phoenix.util.PropertiesUtil;
@@ -269,4 +270,20 @@ public class PhoenixConfigurationUtilTest extends BaseConnectionlessQueryTest {
assertEquals(zkQuorumOverride3, OVERRIDE_CLUSTER_QUORUM);
}
+
+ @Test
+ public void testMrJobTypeOverride() throws Exception {
+ final Job job = Job.getInstance();
+ Configuration configuration = job.getConfiguration();
+ MRJobType mrJobType = PhoenixConfigurationUtil.getMRJobType(configuration,
+ MRJobType.QUERY.name());
+ assertEquals(MRJobType.QUERY.name(), mrJobType.name());
+
+ PhoenixConfigurationUtil.setMRJobType(configuration, MRJobType.UPDATE_STATS);
+ mrJobType = PhoenixConfigurationUtil.getMRJobType(configuration,
+ MRJobType.QUERY.name());
+ assertEquals(MRJobType.UPDATE_STATS.name(), mrJobType.name());
+
+ }
+
}
diff --git a/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/PhoenixPigSchemaUtil.java b/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/PhoenixPigSchemaUtil.java
index d5b870d..7e0203f 100644
--- a/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/PhoenixPigSchemaUtil.java
+++ b/phoenix-pig/src/main/java/org/apache/phoenix/pig/util/PhoenixPigSchemaUtil.java
@@ -57,12 +57,12 @@ public final class PhoenixPigSchemaUtil {
try {
List<ColumnInfo> columns = null;
final SchemaType schemaType = PhoenixConfigurationUtil.getSchemaType(configuration);
- if(SchemaType.QUERY.equals(schemaType)) {
+ if(schemaType == SchemaType.QUERY) {
final String sqlQuery = PhoenixConfigurationUtil.getSelectStatement(configuration);
Preconditions.checkNotNull(sqlQuery, "No Sql Query exists within the configuration");
final SqlQueryToColumnInfoFunction function = new SqlQueryToColumnInfoFunction(configuration);
columns = function.apply(sqlQuery);
- } else {
+ } else if (schemaType == SchemaType.TABLE) {
columns = dependencies.getSelectColumnMetadataList(configuration);
}
ResourceFieldSchema fields[] = new ResourceFieldSchema[columns.size()];
diff --git a/pom.xml b/pom.xml
index 9212185..a172bf8 100644
--- a/pom.xml
+++ b/pom.xml
@@ -607,6 +607,16 @@
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-metrics-api</artifactId>
+ <version>${hbase.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-metrics</artifactId>
+ <version>${hbase.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
<version>${hbase.version}</version>
<type>test-jar</type>