You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ma...@apache.org on 2016/02/24 22:14:06 UTC
[48/50] [abbrv] phoenix git commit: Sync with master branch
Sync with master branch
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/58ec2579
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/58ec2579
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/58ec2579
Branch: refs/heads/calcite
Commit: 58ec2579cd5883bf19b1d24cd3dd9342fec339c8
Parents: 69a2f24 e4acd0c
Author: maryannxue <ma...@gmail.com>
Authored: Tue Feb 23 10:07:19 2016 -0500
Committer: maryannxue <ma...@gmail.com>
Committed: Tue Feb 23 10:07:19 2016 -0500
----------------------------------------------------------------------
bin/queryserver.py | 14 +-
.../phoenix/end2end/AlterTableWithViewsIT.java | 7 +-
.../end2end/ClientTimeArithmeticQueryIT.java | 11 +-
.../phoenix/end2end/CompareDecimalToLongIT.java | 241 --------
.../phoenix/end2end/ContextClassloaderIT.java | 2 +-
.../phoenix/end2end/CsvBulkLoadToolIT.java | 114 ++--
.../org/apache/phoenix/end2end/DateTimeIT.java | 5 +-
.../org/apache/phoenix/end2end/DeleteIT.java | 3 +-
.../apache/phoenix/end2end/DistinctCountIT.java | 2 +-
.../apache/phoenix/end2end/GroupByCaseIT.java | 35 ++
.../org/apache/phoenix/end2end/IndexToolIT.java | 273 +++------
.../phoenix/end2end/LikeExpressionIT.java | 20 +
.../apache/phoenix/end2end/LpadFunctionIT.java | 242 --------
.../org/apache/phoenix/end2end/MapReduceIT.java | 230 ++++++++
.../phoenix/end2end/MultiCfQueryExecIT.java | 51 ++
.../phoenix/end2end/MutableIndexToolIT.java | 128 +++++
.../apache/phoenix/end2end/PercentileIT.java | 2 +-
.../apache/phoenix/end2end/PrimitiveTypeIT.java | 245 ++++++++
.../phoenix/end2end/ProductMetricsIT.java | 6 +-
.../apache/phoenix/end2end/ReverseScanIT.java | 21 +
.../phoenix/end2end/RowValueConstructorIT.java | 4 +-
.../end2end/StatsCollectionDisabledIT.java | 70 +++
.../phoenix/end2end/StatsCollectorIT.java | 28 +-
.../StatsCollectorWithSplitsAndMultiCFIT.java | 69 ++-
.../org/apache/phoenix/end2end/StringIT.java | 254 +++++++++
.../org/apache/phoenix/end2end/UnionAllIT.java | 49 +-
.../phoenix/end2end/UserDefinedFunctionsIT.java | 8 +-
.../phoenix/end2end/VariableLengthPKIT.java | 2 +-
.../end2end/index/DropIndexDuringUpsertIT.java | 2 +-
.../end2end/index/IndexExpressionIT.java | 7 +-
.../apache/phoenix/end2end/index/IndexIT.java | 5 +-
.../end2end/index/MutableIndexFailureIT.java | 398 +++++--------
.../index/MutableIndexReplicationIT.java | 2 +-
.../end2end/index/ReadOnlyIndexFailureIT.java | 284 ++++++++++
.../salted/SaltedTableUpsertSelectIT.java | 57 ++
.../EndToEndCoveredColumnsIndexBuilderIT.java | 2 +-
.../example/EndToEndCoveredIndexingIT.java | 5 +-
.../org/apache/phoenix/tx/TransactionIT.java | 12 +-
phoenix-core/src/main/antlr3/PhoenixSQL.g | 5 +-
.../apache/phoenix/calcite/PhoenixSchema.java | 2 +-
.../apache/phoenix/calcite/PhoenixTable.java | 12 +-
.../calcite/rel/PhoenixRelImplementorImpl.java | 2 +-
.../phoenix/calcite/rel/PhoenixTableScan.java | 5 +-
.../apache/phoenix/compile/DeleteCompiler.java | 6 +-
.../apache/phoenix/compile/FromCompiler.java | 2 +-
.../apache/phoenix/compile/GroupByCompiler.java | 74 ++-
.../apache/phoenix/compile/JoinCompiler.java | 2 +-
.../apache/phoenix/compile/QueryCompiler.java | 3 +-
.../org/apache/phoenix/compile/ScanRanges.java | 2 +
.../compile/TupleProjectionCompiler.java | 4 +-
.../apache/phoenix/compile/UnionCompiler.java | 2 +-
.../apache/phoenix/compile/UpsertCompiler.java | 7 +-
.../coprocessor/BaseScannerRegionObserver.java | 16 +-
.../coprocessor/DelegateRegionObserver.java | 562 +++++++++++++++++++
.../coprocessor/HashJoinRegionScanner.java | 71 ++-
.../coprocessor/MetaDataEndpointImpl.java | 92 +--
.../coprocessor/MetaDataRegionObserver.java | 136 ++++-
.../PhoenixTransactionalProcessor.java | 28 +
.../UngroupedAggregateRegionObserver.java | 31 +-
.../generated/PGuidePostsProtos.java | 336 ++++++++++-
.../coprocessor/generated/PTableProtos.java | 103 +++-
.../phoenix/exception/SQLExceptionCode.java | 7 +-
.../apache/phoenix/execute/AggregatePlan.java | 16 +-
.../apache/phoenix/execute/MutationState.java | 39 +-
.../org/apache/phoenix/execute/ScanPlan.java | 19 +-
.../phoenix/expression/InListExpression.java | 2 +-
.../phoenix/expression/LiteralExpression.java | 2 +-
.../expression/ProjectedColumnExpression.java | 11 +-
.../expression/StringConcatExpression.java | 21 +-
.../expression/util/regex/JavaPattern.java | 2 +-
.../visitor/CloneExpressionVisitor.java | 2 +-
.../phoenix/filter/ColumnProjectionFilter.java | 2 +
.../apache/phoenix/filter/SkipScanFilter.java | 47 +-
.../hbase/index/builder/IndexBuildManager.java | 10 +-
.../hbase/index/covered/LocalTableState.java | 19 +-
.../phoenix/hbase/index/covered/TableState.java | 7 +-
.../index/covered/data/LocalHBaseState.java | 6 +-
.../hbase/index/covered/data/LocalTable.java | 9 +-
.../example/CoveredColumnIndexCodec.java | 4 +-
.../hbase/index/scanner/ScannerBuilder.java | 1 -
.../index/write/DelegateIndexFailurePolicy.java | 58 ++
.../apache/phoenix/index/IndexMaintainer.java | 4 +-
.../phoenix/index/PhoenixIndexBuilder.java | 4 +-
.../apache/phoenix/index/PhoenixIndexCodec.java | 12 +-
.../index/PhoenixIndexFailurePolicy.java | 48 +-
.../phoenix/index/PhoenixIndexMetaData.java | 10 +-
.../index/PhoenixTransactionalIndexer.java | 7 +-
.../phoenix/iterate/BaseResultIterators.java | 293 +++++-----
.../apache/phoenix/iterate/ExplainTable.java | 32 +-
.../apache/phoenix/iterate/SerialIterators.java | 3 +-
.../apache/phoenix/jdbc/PhoenixConnection.java | 7 +-
.../org/apache/phoenix/jdbc/PhoenixDriver.java | 32 +-
.../phoenix/mapreduce/AbstractBulkLoadTool.java | 22 +-
.../mapreduce/FormatToKeyValueMapper.java | 164 +++++-
.../mapreduce/FormatToKeyValueReducer.java | 127 ++++-
.../phoenix/mapreduce/PhoenixInputFormat.java | 11 +-
.../mapreduce/bulkload/TableRowkeyPair.java | 45 +-
.../mapreduce/bulkload/TargetTableRef.java | 2 +-
.../bulkload/TargetTableRefFunctions.java | 22 +-
.../phoenix/mapreduce/index/IndexTool.java | 9 +-
.../index/PhoenixIndexImportDirectMapper.java | 7 +-
.../index/PhoenixIndexImportMapper.java | 6 +-
.../util/PhoenixConfigurationUtil.java | 6 +-
.../mapreduce/util/PhoenixMapReduceUtil.java | 68 +--
.../java/org/apache/phoenix/parse/HintNode.java | 4 +
.../apache/phoenix/parse/NamedTableNode.java | 8 +
.../phoenix/query/ConnectionQueryServices.java | 1 +
.../query/ConnectionQueryServicesImpl.java | 96 ++--
.../query/ConnectionlessQueryServicesImpl.java | 15 +
.../query/DelegateConnectionQueryServices.java | 6 +
.../org/apache/phoenix/query/QueryServices.java | 4 +
.../phoenix/query/QueryServicesOptions.java | 3 +-
.../apache/phoenix/schema/DelegateTable.java | 5 +
.../apache/phoenix/schema/MetaDataClient.java | 39 +-
.../java/org/apache/phoenix/schema/PTable.java | 1 +
.../org/apache/phoenix/schema/PTableImpl.java | 76 +--
.../stats/DefaultStatisticsCollector.java | 231 ++++++++
.../phoenix/schema/stats/GuidePostsInfo.java | 69 ++-
.../schema/stats/GuidePostsInfoBuilder.java | 61 +-
.../schema/stats/NoOpStatisticsCollector.java | 72 +++
.../phoenix/schema/stats/PTableStats.java | 2 +-
.../schema/stats/StatisticsCollector.java | 222 +-------
.../stats/StatisticsCollectorFactory.java | 63 +++
.../phoenix/schema/stats/StatisticsScanner.java | 7 +-
.../phoenix/schema/stats/StatisticsWriter.java | 24 +-
.../apache/phoenix/schema/types/PDataType.java | 3 +
.../apache/phoenix/schema/types/PTimestamp.java | 407 ++++++++------
.../java/org/apache/phoenix/util/DateUtil.java | 2 +-
.../java/org/apache/phoenix/util/IndexUtil.java | 16 +-
.../org/apache/phoenix/util/PhoenixRuntime.java | 8 +-
.../java/org/apache/phoenix/util/ScanUtil.java | 49 +-
.../org/apache/phoenix/util/UpgradeUtil.java | 16 +-
.../phoenix/compile/QueryCompilerTest.java | 65 +++
.../compile/StatementHintsCompilationTest.java | 21 +-
.../phoenix/compile/WhereOptimizerTest.java | 5 +-
.../phoenix/execute/CorrelatePlanTest.java | 2 +-
.../index/covered/TestLocalTableState.java | 10 +-
.../example/TestCoveredColumnIndexCodec.java | 4 +-
.../mapreduce/bulkload/TestTableRowkeyPair.java | 67 +++
.../apache/phoenix/parse/QueryParserTest.java | 6 +
.../java/org/apache/phoenix/query/BaseTest.java | 65 ++-
.../org/apache/phoenix/query/QueryPlanTest.java | 24 +-
.../phoenix/schema/types/PDataTypeTest.java | 60 ++
.../java/org/apache/phoenix/util/TestUtil.java | 7 +-
.../apache/phoenix/pherf/PherfConstants.java | 2 +-
.../phoenix/pherf/rules/RulesApplier.java | 17 +-
.../apache/phoenix/pherf/RuleGeneratorTest.java | 5 +-
.../test/resources/scenario/test_scenario.xml | 6 +-
phoenix-protocol/src/main/PGuidePosts.proto | 2 +
phoenix-protocol/src/main/PTable.proto | 1 +
pom.xml | 5 +-
151 files changed, 5293 insertions(+), 2229 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
index fd630ab,0000000..b8401c6
mode 100644,000000..100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixSchema.java
@@@ -1,331 -1,0 +1,331 @@@
+package org.apache.phoenix.calcite;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import org.apache.calcite.jdbc.CalciteSchema;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.materialize.MaterializationService;
+import org.apache.calcite.schema.*;
+import org.apache.calcite.schema.impl.ViewTable;
+import org.apache.phoenix.compile.ColumnResolver;
+import org.apache.phoenix.compile.FromCompiler;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.parse.ColumnDef;
+import org.apache.phoenix.parse.NamedTableNode;
+import org.apache.phoenix.parse.TableName;
+import org.apache.phoenix.schema.MetaDataClient;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTable.ViewType;
+import org.apache.phoenix.schema.PTableImpl;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.TableNotFoundException;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.SchemaUtil;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+/**
+ * Implementation of Calcite's {@link Schema} SPI for Phoenix.
+ *
+ * TODO
+ * 1) change this to non-caching mode??
+ * 2) how to deal with define indexes and views since they require a CalciteSchema
+ * instance??
+ *
+ */
+public class PhoenixSchema implements Schema {
+ public static final Factory FACTORY = new Factory();
+
+ public final PhoenixConnection pc;
+
+ protected final String name;
+ protected final String schemaName;
+ protected final SchemaPlus parentSchema;
+ protected final MetaDataClient client;
+
+ protected final Map<String, Schema> subSchemas;
+ protected final Map<String, Table> tables;
+ protected final Map<String, Function> views;
+ protected final Set<PTable> viewTables;
+
+ protected PhoenixSchema(String name, String schemaName,
+ SchemaPlus parentSchema, PhoenixConnection pc) {
+ this.name = name;
+ this.schemaName = schemaName;
+ this.parentSchema = parentSchema;
+ this.pc = pc;
+ this.client = new MetaDataClient(pc);
+ this.subSchemas = Maps.newHashMap();
+ this.tables = Maps.newHashMap();
+ this.views = Maps.newHashMap();
+ this.viewTables = Sets.newHashSet();
+ }
+
+ private static Schema create(SchemaPlus parentSchema,
+ String name, Map<String, Object> operand) {
+ String url = (String) operand.get("url");
+ final Properties properties = new Properties();
+ for (Map.Entry<String, Object> entry : operand.entrySet()) {
+ properties.setProperty(entry.getKey(), String.valueOf(entry.getValue()));
+ }
+ try {
+ Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
+ final Connection connection =
+ DriverManager.getConnection(url, properties);
+ final PhoenixConnection phoenixConnection =
+ connection.unwrap(PhoenixConnection.class);
+ return new PhoenixSchema(name, null, parentSchema, phoenixConnection);
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException(e);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public Table getTable(String name) {
+ Table table = tables.get(name);
+ if (table != null) {
+ return table;
+ }
+
+ try {
+ ColumnResolver x = FromCompiler.getResolver(
+ NamedTableNode.create(
+ null,
+ TableName.create(schemaName, name),
+ ImmutableList.<ColumnDef>of()), pc);
+ final List<TableRef> tables = x.getTables();
+ assert tables.size() == 1;
+ PTable pTable = tables.get(0).getTable();
+ if (!isView(pTable)) {
+ pTable = fixTableMultiTenancy(pTable);
+ table = new PhoenixTable(pc, pTable);
+ }
+ } catch (TableNotFoundException e) {
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+
+ if (table == null) {
+ table = resolveSequence(name);
+ }
+
+ if (table != null) {
+ tables.put(name, table);
+ }
+ return table;
+ }
+
+ @Override
+ public Set<String> getTableNames() {
+ return tables.keySet();
+ }
+
+ @Override
+ public Collection<Function> getFunctions(String name) {
+ Function func = views.get(name);
+ if (func != null) {
+ return ImmutableList.of(func);
+ }
+
+ try {
+ ColumnResolver x = FromCompiler.getResolver(
+ NamedTableNode.create(
+ null,
+ TableName.create(schemaName, name),
+ ImmutableList.<ColumnDef>of()), pc);
+ final List<TableRef> tables = x.getTables();
+ assert tables.size() == 1;
+ PTable pTable = tables.get(0).getTable();
+ if (isView(pTable)) {
+ String viewSql = pTable.getViewStatement();
+ if (viewSql == null) {
+ viewSql = "select * from "
+ + SchemaUtil.getEscapedFullTableName(
+ pTable.getPhysicalName().getString());
+ }
+ SchemaPlus schema = parentSchema.getSubSchema(this.name);
+ SchemaPlus viewSqlSchema =
+ this.schemaName == null ? schema : parentSchema;
+ func = ViewTable.viewMacro(schema, viewSql,
+ CalciteSchema.from(viewSqlSchema).path(null),
+ pTable.getViewType() == ViewType.UPDATABLE);
+ views.put(name, func);
+ viewTables.add(pTable);
+ }
+ } catch (TableNotFoundException e) {
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+
+ return func == null ? Collections.<Function>emptyList() : ImmutableList.of(func);
+ }
+
+ @Override
+ public Set<String> getFunctionNames() {
+ return views.keySet();
+ }
+
+ @Override
+ public Schema getSubSchema(String name) {
+ if (schemaName != null) {
+ return null;
+ }
+
+ Schema schema = subSchemas.get(name);
+ if (schema != null) {
+ return schema;
+ }
+
+ schema = new PhoenixSchema(name, name, parentSchema.getSubSchema(this.name), pc);
+ subSchemas.put(name, schema);
+ return schema;
+ }
+
+ @Override
+ public Set<String> getSubSchemaNames() {
+ return subSchemas.keySet();
+ }
+
+ @Override
+ public Expression getExpression(SchemaPlus parentSchema, String name) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isMutable() {
+ return true;
+ }
+
+ @Override
+ public boolean contentsHaveChangedSince(long lastCheck, long now) {
+ return false;
+ }
+
+ public void defineIndexesAsMaterializations() {
+ SchemaPlus schema = parentSchema.getSubSchema(this.name);
+ SchemaPlus viewSqlSchema =
+ this.schemaName == null ? schema : parentSchema;
+ CalciteSchema calciteSchema = CalciteSchema.from(schema);
+ List<String> path = CalciteSchema.from(viewSqlSchema).path(null);
+ try {
+ for (Table table : tables.values()) {
+ if (table instanceof PhoenixTable) {
+ PTable pTable = ((PhoenixTable) table).pTable;
+ for (PTable index : pTable.getIndexes()) {
+ addMaterialization(index, path, calciteSchema);
+ }
+ }
+ }
+ for (PTable pTable : viewTables) {
+ for (PTable index : pTable.getIndexes()) {
+ if (index.getParentName().equals(pTable.getName())) {
+ addMaterialization(index, path, calciteSchema);
+ }
+ }
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void addMaterialization(PTable index, List<String> path,
+ CalciteSchema calciteSchema) throws SQLException {
+ index = fixTableMultiTenancy(index);
+ StringBuffer sb = new StringBuffer();
+ sb.append("SELECT");
+ for (PColumn column : PhoenixTable.getMappedColumns(index)) {
+ String indexColumnName = column.getName().getString();
+ String dataColumnName = IndexUtil.getDataColumnName(indexColumnName);
+ sb.append(",").append(SchemaUtil.getEscapedFullColumnName(dataColumnName));
+ sb.append(" ").append(SchemaUtil.getEscapedFullColumnName(indexColumnName));
+ }
+ sb.setCharAt(6, ' '); // replace first comma with space.
+ sb.append(" FROM ").append(SchemaUtil.getEscapedFullTableName(index.getParentName().getString()));
+ MaterializationService.instance().defineMaterialization(
+ calciteSchema, null, sb.toString(), path, index.getTableName().getString(), true, true);
+ }
+
+ private boolean isView(PTable table) {
+ return table.getType() == PTableType.VIEW
+ && table.getViewType() != ViewType.MAPPED;
+ }
+
+ private PTable fixTableMultiTenancy(PTable table) throws SQLException {
+ if (pc.getTenantId() != null || !table.isMultiTenant()) {
+ return table;
+ }
+ return PTableImpl.makePTable(
+ table.getTenantId(), table.getSchemaName(), table.getTableName(), table.getType(), table.getIndexState(), table.getTimeStamp(),
+ table.getSequenceNumber(), table.getPKName(), table.getBucketNum(), PTableImpl.getColumnsToClone(table), table.getParentSchemaName(), table.getParentTableName(),
+ table.getIndexes(), table.isImmutableRows(), table.getPhysicalNames(), table.getDefaultFamilyName(), table.getViewStatement(),
+ table.isWALDisabled(), false, table.getStoreNulls(), table.getViewType(), table.getViewIndexId(), table.getIndexType(),
- table.rowKeyOrderOptimizable(), table.isTransactional(), table.getUpdateCacheFrequency(), table.getTableStats(), table.getBaseColumnCount());
++ table.rowKeyOrderOptimizable(), table.isTransactional(), table.getUpdateCacheFrequency(), table.getTableStats(), table.getBaseColumnCount(), table.getIndexDisableTimestamp());
+ }
+
+ private PhoenixSequence resolveSequence(String name) {
+ try {
+ // FIXME: Do this the same way as resolving a table after PHOENIX-2489.
+ String tenantId = pc.getTenantId() == null ? null : pc.getTenantId().getString();
+ String q = "select 1 from " + PhoenixDatabaseMetaData.SEQUENCE_FULLNAME_ESCAPED
+ + " where " + PhoenixDatabaseMetaData.SEQUENCE_SCHEMA
+ + (schemaName == null ? " is null" : " = '" + schemaName + "'")
+ + " and " + PhoenixDatabaseMetaData.SEQUENCE_NAME
+ + " = '" + name + "'"
+ + " and " + PhoenixDatabaseMetaData.TENANT_ID
+ + (tenantId == null ? " is null" : " = '" + tenantId + "'");
+ ResultSet rs = pc.createStatement().executeQuery(q);
+ if (rs.next()) {
+ return new PhoenixSequence(schemaName, name, pc);
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+
+ return null;
+ }
+
+ /** Schema factory that creates a
+ * {@link org.apache.phoenix.calcite.PhoenixSchema}.
+ * This allows you to create a Phoenix schema inside a model.json file.
+ *
+ * <pre>{@code
+ * {
+ * version: '1.0',
+ * defaultSchema: 'HR',
+ * schemas: [
+ * {
+ * name: 'HR',
+ * type: 'custom',
+ * factory: 'org.apache.phoenix.calcite.PhoenixSchema.Factory',
+ * operand: {
+ * url: "jdbc:phoenix:localhost",
+ * user: "scott",
+ * password: "tiger"
+ * }
+ * }
+ * ]
+ * }
+ * }</pre>
+ */
+ public static class Factory implements SchemaFactory {
+ public Schema create(SchemaPlus parentSchema, String name, Map<String, Object> operand) {
+ return PhoenixSchema.create(parentSchema, name, operand);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixTable.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixTable.java
index fea4c8a,0000000..101ee5d
mode 100644,000000..100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixTable.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/PhoenixTable.java
@@@ -1,206 -1,0 +1,210 @@@
+package org.apache.phoenix.calcite;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelDistribution;
+import org.apache.calcite.rel.RelDistributions;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.RelFieldCollation.Direction;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.Statistic;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.schema.impl.AbstractTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.phoenix.calcite.rel.PhoenixTableScan;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PColumnFamily;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.RowKeySchema;
+import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.stats.GuidePostsInfo;
+import org.apache.phoenix.schema.stats.StatisticsUtil;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.SizedUtil;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+/**
+ * Implementation of Calcite {@link org.apache.calcite.schema.Table} SPI for
+ * Phoenix.
+ */
+public class PhoenixTable extends AbstractTable implements TranslatableTable {
+ public final PTable pTable;
+ public final List<PColumn> mappedColumns;
+ public final ImmutableBitSet pkBitSet;
+ public final RelCollation collation;
+ public final long byteCount;
+ public final long rowCount;
+ public final PhoenixConnection pc;
+
+ public static List<PColumn> getMappedColumns(PTable pTable) {
+ if (pTable.getBucketNum() == null
+ && !pTable.isMultiTenant()
+ && pTable.getViewIndexId() == null) {
+ return pTable.getColumns();
+ }
+
+ List<PColumn> columns = Lists.newArrayList(pTable.getColumns());
+ if (pTable.getViewIndexId() != null) {
+ columns.remove((pTable.getBucketNum() == null ? 0 : 1) + (pTable.isMultiTenant() ? 1 : 0));
+ }
+ if (pTable.isMultiTenant()) {
+ columns.remove(pTable.getBucketNum() == null ? 0 : 1);
+ }
+ if (pTable.getBucketNum() != null) {
+ columns.remove(0);
+ }
+ return columns;
+ }
+
+ public PhoenixTable(PhoenixConnection pc, PTable pTable) {
+ this.pc = Preconditions.checkNotNull(pc);
+ this.pTable = Preconditions.checkNotNull(pTable);
+ this.mappedColumns = getMappedColumns(pTable);
+ List<Integer> pkPositions = Lists.<Integer> newArrayList();
+ List<RelFieldCollation> fieldCollations = Lists.<RelFieldCollation> newArrayList();
+ for (int i = 0; i < mappedColumns.size(); i++) {
+ PColumn column = mappedColumns.get(i);
+ if (SchemaUtil.isPKColumn(column)) {
+ SortOrder sortOrder = column.getSortOrder();
+ pkPositions.add(i);
+ fieldCollations.add(new RelFieldCollation(i, sortOrder == SortOrder.ASC ? Direction.ASCENDING : Direction.DESCENDING));
+ }
+ }
+ this.pkBitSet = ImmutableBitSet.of(pkPositions);
+ this.collation = RelCollationTraitDef.INSTANCE.canonize(RelCollations.of(fieldCollations));
+ byte[] emptyCf = SchemaUtil.getEmptyColumnFamily(pTable);
+ GuidePostsInfo info = pTable.getTableStats().getGuidePosts().get(emptyCf);
- long rowCount;
- long byteCount;
++ long rowCount = 0;
++ long byteCount = 0;
+ try {
+ if (info == null) {
+ // TODO The props might not be the same as server props.
+ int guidepostPerRegion = pc.getQueryServices().getProps().getInt(
+ QueryServices.STATS_GUIDEPOST_PER_REGION_ATTRIB,
+ QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_PER_REGION);
+ long guidepostWidth = pc.getQueryServices().getProps().getLong(
+ QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB,
+ QueryServicesOptions.DEFAULT_STATS_GUIDEPOST_WIDTH_BYTES);
+ HTableDescriptor desc = null;
+ if (guidepostPerRegion > 0) {
+ desc = pc.getQueryServices().getAdmin().getTableDescriptor(
+ pTable.getPhysicalName().getBytes());
+ }
+ byteCount = StatisticsUtil.getGuidePostDepth(
+ guidepostPerRegion, guidepostWidth, desc) / 2;
+ long keySize = pTable.getRowKeySchema().getEstimatedByteSize();
+ long rowSize = 0;
+ for (PColumnFamily cf : pTable.getColumnFamilies()) {
+ for (PColumn column : cf.getColumns()) {
+ Integer maxLength = column.getMaxLength();
+ int byteSize = column.getDataType().isFixedWidth() ?
+ maxLength == null ?
+ column.getDataType().getByteSize()
+ : maxLength
+ : RowKeySchema.ESTIMATED_VARIABLE_LENGTH_SIZE;
+ rowSize += SizedUtil.KEY_VALUE_SIZE + keySize + byteSize;
+ }
+ }
+ if (rowSize == 0) {
+ rowSize = keySize;
+ }
+ rowCount = byteCount / rowSize;
+ } else {
- byteCount = info.getByteCount();
- rowCount = info.getRowCount();
++ for (long b : info.getByteCounts()) {
++ byteCount += b;
++ }
++ for (long r : info.getRowCounts()) {
++ rowCount += r;
++ }
+ }
+ } catch (SQLException | IOException e) {
+ throw new RuntimeException(e);
+ }
+ this.byteCount = byteCount;
+ this.rowCount = rowCount;
+ }
+
+ public PTable getTable() {
+ return pTable;
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public RelDataType getRowType(RelDataTypeFactory typeFactory) {
+ final RelDataTypeFactory.FieldInfoBuilder builder = typeFactory.builder();
+ for (int i = 0; i < mappedColumns.size(); i++) {
+ PColumn pColumn = mappedColumns.get(i);
+ final PDataType baseType =
+ pColumn.getDataType().isArrayType() ?
+ PDataType.fromTypeId(pColumn.getDataType().getSqlType() - PDataType.ARRAY_TYPE_BASE)
+ : pColumn.getDataType();
+ final int sqlTypeId = baseType.getResultSetSqlType();
+ final PDataType pDataType = PDataType.fromTypeId(sqlTypeId);
+ final SqlTypeName sqlTypeName1 = SqlTypeName.valueOf(pDataType.getSqlTypeName());
+ final Integer maxLength = pColumn.getMaxLength();
+ final Integer scale = pColumn.getScale();
+ RelDataType type;
+ if (maxLength != null && scale != null) {
+ type = typeFactory.createSqlType(sqlTypeName1, maxLength, scale);
+ } else if (maxLength != null) {
+ type = typeFactory.createSqlType(sqlTypeName1, maxLength);
+ } else {
+ type = typeFactory.createSqlType(sqlTypeName1);
+ }
+ if (pColumn.getDataType().isArrayType()) {
+ final Integer arraySize = pColumn.getArraySize();
+ type = typeFactory.createArrayType(type, arraySize == null ? -1 : arraySize);
+ }
+ builder.add(pColumn.getName().getString(), type);
+ builder.nullable(pColumn.isNullable());
+ }
+ return builder.build();
+ }
+
+ @Override
+ public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {
+ return PhoenixTableScan.create(context.getCluster(), relOptTable);
+ }
+
+ @Override
+ public Statistic getStatistic() {
+ return new Statistic() {
+ @Override
+ public Double getRowCount() {
+ return (double) rowCount;
+ }
+
+ @Override
+ public boolean isKey(ImmutableBitSet immutableBitSet) {
+ return immutableBitSet.contains(pkBitSet);
+ }
+
+ @Override
+ public List<RelCollation> getCollations() {
+ return ImmutableList.<RelCollation>of(collation);
+ }
+
+ @Override
+ public RelDistribution getDistribution() {
+ return RelDistributions.RANDOM_DISTRIBUTED;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixRelImplementorImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixRelImplementorImpl.java
index e1afc14,0000000..a33cd18
mode 100644,000000..100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixRelImplementorImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixRelImplementorImpl.java
@@@ -1,191 -1,0 +1,191 @@@
+package org.apache.phoenix.calcite.rel;
+
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Stack;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.calcite.CalciteUtils;
+import org.apache.phoenix.calcite.PhoenixSequence;
+import org.apache.phoenix.calcite.PhoenixTable;
+import org.apache.phoenix.calcite.rel.PhoenixRel.ImplementorContext;
+import org.apache.phoenix.compile.ColumnProjector;
+import org.apache.phoenix.compile.ExpressionProjector;
+import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.SequenceManager;
+import org.apache.phoenix.compile.SequenceValueExpression;
+import org.apache.phoenix.compile.TupleProjectionCompiler;
+import org.apache.phoenix.coprocessor.MetaDataProtocol;
+import org.apache.phoenix.execute.RuntimeContext;
+import org.apache.phoenix.execute.TupleProjector;
+import org.apache.phoenix.expression.ColumnExpression;
+import org.apache.phoenix.expression.CorrelateVariableFieldAccessExpression;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.ParseNodeFactory;
+import org.apache.phoenix.parse.SequenceValueParseNode;
+import org.apache.phoenix.parse.TableName;
+import org.apache.phoenix.schema.ColumnRef;
+import org.apache.phoenix.schema.KeyValueSchema;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PColumnImpl;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PNameFactory;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableImpl;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.KeyValueSchema.KeyValueSchemaBuilder;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.SchemaUtil;
+
+import com.google.common.collect.Lists;
+
+public class PhoenixRelImplementorImpl implements PhoenixRel.Implementor {
+ private final RuntimeContext runtimeContext;
+ private TableRef tableRef;
+ private List<PColumn> mappedColumns;
+ private Stack<ImplementorContext> contextStack;
+ private SequenceManager sequenceManager;
+
+ public PhoenixRelImplementorImpl(RuntimeContext runtimeContext) {
+ this.runtimeContext = runtimeContext;
+ this.contextStack = new Stack<ImplementorContext>();
+ }
+
+ @Override
+ public QueryPlan visitInput(int i, PhoenixRel input) {
+ return input.implement(this);
+ }
+
+ @Override
+ public ColumnExpression newColumnExpression(int index) {
+ ColumnRef colRef = new ColumnRef(this.tableRef, this.mappedColumns.get(index).getPosition());
+ return colRef.newColumnExpression();
+ }
+
+ @SuppressWarnings("rawtypes")
+ @Override
+ public Expression newFieldAccessExpression(String variableId, int index, PDataType type) {
+ Expression fieldAccessExpr = runtimeContext.newCorrelateVariableReference(variableId, index);
+ return new CorrelateVariableFieldAccessExpression(runtimeContext, variableId, fieldAccessExpr);
+ }
+
+ @Override
+ public SequenceValueExpression newSequenceExpression(PhoenixSequence seq, SequenceValueParseNode.Op op) {
+ PName tenantName = seq.pc.getTenantId();
+ TableName tableName = TableName.create(seq.schemaName, seq.sequenceName);
+ try {
+ return sequenceManager.newSequenceReference(tenantName, tableName, null, op);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public RuntimeContext getRuntimeContext() {
+ return runtimeContext;
+ }
+
+ @Override
+ public void setTableRef(TableRef tableRef) {
+ this.tableRef = tableRef;
+ this.mappedColumns = PhoenixTable.getMappedColumns(tableRef.getTable());
+ }
+
+ @Override
+ public TableRef getTableRef() {
+ return this.tableRef;
+ }
+
+ @Override
+ public void setSequenceManager(SequenceManager sequenceManager) {
+ this.sequenceManager = sequenceManager;
+ }
+
+ @Override
+ public void pushContext(ImplementorContext context) {
+ this.contextStack.push(context);
+ }
+
+ @Override
+ public ImplementorContext popContext() {
+ return contextStack.pop();
+ }
+
+ @Override
+ public ImplementorContext getCurrentContext() {
+ return contextStack.peek();
+ }
+
+ @Override
+ public PTable createProjectedTable() {
+ List<ColumnRef> sourceColumnRefs = Lists.<ColumnRef> newArrayList();
+ List<PColumn> columns = getCurrentContext().retainPKColumns ?
+ getTableRef().getTable().getColumns() : mappedColumns;
+ for (PColumn column : columns) {
+ sourceColumnRefs.add(new ColumnRef(getTableRef(), column.getPosition()));
+ }
+
+ try {
+ return TupleProjectionCompiler.createProjectedTable(getTableRef(), sourceColumnRefs, getCurrentContext().retainPKColumns);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public TupleProjector createTupleProjector() {
+ KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0);
+ List<Expression> exprs = Lists.<Expression> newArrayList();
+ for (PColumn column : mappedColumns) {
+ if (!SchemaUtil.isPKColumn(column) || !getCurrentContext().retainPKColumns) {
+ Expression expr = new ColumnRef(tableRef, column.getPosition()).newColumnExpression();
+ exprs.add(expr);
+ builder.addField(expr);
+ }
+ }
+
+ return new TupleProjector(builder.build(), exprs.toArray(new Expression[exprs.size()]));
+ }
+
+ @Override
+ public RowProjector createRowProjector() {
+ List<ColumnProjector> columnProjectors = Lists.<ColumnProjector>newArrayList();
+ for (int i = 0; i < mappedColumns.size(); i++) {
+ PColumn column = mappedColumns.get(i);
+ Expression expr = newColumnExpression(i); // Do not use column.position() here.
+ columnProjectors.add(new ExpressionProjector(column.getName().getString(), getTableRef().getTable().getName().getString(), expr, false));
+ }
+ // TODO get estimate row size
+ return new RowProjector(columnProjectors, 0, false);
+ }
+
+ @Override
+ public TupleProjector project(List<Expression> exprs) {
+ KeyValueSchema.KeyValueSchemaBuilder builder = new KeyValueSchema.KeyValueSchemaBuilder(0);
+ List<PColumn> columns = Lists.<PColumn>newArrayList();
+ for (int i = 0; i < exprs.size(); i++) {
+ String name = ParseNodeFactory.createTempAlias();
+ Expression expr = exprs.get(i);
+ builder.addField(expr);
+ columns.add(new PColumnImpl(PNameFactory.newName(name), PNameFactory.newName(TupleProjector.VALUE_COLUMN_FAMILY),
+ expr.getDataType(), expr.getMaxLength(), expr.getScale(), expr.isNullable(),
+ i, expr.getSortOrder(), null, null, false, name, false, false));
+ }
+ try {
+ PTable pTable = PTableImpl.makePTable(null, PName.EMPTY_NAME, PName.EMPTY_NAME,
+ PTableType.SUBQUERY, null, MetaDataProtocol.MIN_TABLE_TIMESTAMP, PTable.INITIAL_SEQ_NUM,
+ null, null, columns, null, null, Collections.<PTable>emptyList(),
+ false, Collections.<PName>emptyList(), null, null, false, false, false, null,
- null, null, true, false, 0);
++ null, null, true, false, 0, 0);
+ this.setTableRef(new TableRef(CalciteUtils.createTempAlias(), pTable, HConstants.LATEST_TIMESTAMP, false));
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+
+ return new TupleProjector(builder.build(), exprs.toArray(new Expression[exprs.size()]));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixTableScan.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixTableScan.java
index 522cc7b,0000000..6dbe5b1
mode 100644,000000..100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixTableScan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/calcite/rel/PhoenixTableScan.java
@@@ -1,360 -1,0 +1,363 @@@
+package org.apache.phoenix.calcite.rel;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelOptUtil.InputFinder;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelCollation;
+import org.apache.calcite.rel.RelCollationTraitDef;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.phoenix.calcite.CalciteUtils;
+import org.apache.phoenix.calcite.PhoenixTable;
+import org.apache.phoenix.compile.ColumnResolver;
+import org.apache.phoenix.compile.FromCompiler;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.compile.SequenceManager;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.compile.WhereCompiler;
+import org.apache.phoenix.compile.WhereOptimizer;
+import org.apache.phoenix.execute.ScanPlan;
+import org.apache.phoenix.execute.TupleProjector;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.LiteralExpression;
+import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.iterate.ParallelIteratorFactory;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.parse.SelectStatement;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.stats.GuidePostsInfo;
+import org.apache.phoenix.schema.stats.StatisticsUtil;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.util.SchemaUtil;
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableList.Builder;
+
+/**
+ * Scan of a Phoenix table.
+ */
+public class PhoenixTableScan extends TableScan implements PhoenixRel {
+ public enum ScanOrder {
+ NONE,
+ FORWARD,
+ REVERSE,
+ }
+
+ public final RexNode filter;
+ public final ScanOrder scanOrder;
+ public final ScanRanges scanRanges;
+
+ protected final GuidePostsInfo filteredGuideposts;
+ protected final float rowCountFactor;
+
+ public static PhoenixTableScan create(RelOptCluster cluster, final RelOptTable table) {
+ return create(cluster, table, null,
+ getDefaultScanOrder(table.unwrap(PhoenixTable.class)));
+ }
+
+ public static PhoenixTableScan create(RelOptCluster cluster, final RelOptTable table,
+ RexNode filter, final ScanOrder scanOrder) {
+ final RelTraitSet traits =
+ cluster.traitSetOf(PhoenixConvention.SERVER)
+ .replaceIfs(RelCollationTraitDef.INSTANCE,
+ new Supplier<List<RelCollation>>() {
+ public List<RelCollation> get() {
+ if (scanOrder == ScanOrder.NONE) {
+ return ImmutableList.of();
+ }
+ List<RelCollation> collations = table.getCollationList();
+ return scanOrder == ScanOrder.FORWARD ? collations : reverse(collations);
+ }
+ });
+ return new PhoenixTableScan(cluster, traits, table, filter, scanOrder);
+ }
+
+ private PhoenixTableScan(RelOptCluster cluster, RelTraitSet traits, RelOptTable table, RexNode filter, ScanOrder scanOrder) {
+ super(cluster, traits, table);
+ this.filter = filter;
+ this.scanOrder = scanOrder;
+ final PhoenixTable phoenixTable = table.unwrap(PhoenixTable.class);
+ this.rowCountFactor = phoenixTable.pc.getQueryServices()
+ .getProps().getFloat(PhoenixRel.ROW_COUNT_FACTOR, 1f);
+
+ ScanRanges scanRanges = null;
+ GuidePostsInfo info = null;
+ HTableInterface statsHTable = null;
+ if (filter != null) {
+ try {
+ // TODO simplify this code
+ PTable pTable = phoenixTable.getTable();
+ TableRef tableRef = new TableRef(CalciteUtils.createTempAlias(), pTable, HConstants.LATEST_TIMESTAMP, false);
+ // We use a implementor with a special implementation for field access
+ // here, which translates RexFieldAccess into a LiteralExpression
+ // with a sample value. This will achieve 3 goals at a time:
+ // 1) avoid getting exception when translating RexFieldAccess at this
+ // time when the correlate variable has not been defined yet.
+ // 2) get a guess of ScanRange even if the runtime value is absent.
+ // 3) test whether this dynamic filter is worth a recompile at runtime.
+ Implementor tmpImplementor = new PhoenixRelImplementorImpl(null) {
+ @SuppressWarnings("rawtypes")
+ @Override
+ public Expression newFieldAccessExpression(String variableId, int index, PDataType type) {
+ try {
+ return LiteralExpression.newConstant(type.getSampleValue(), type);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ tmpImplementor.setTableRef(tableRef);
+ SelectStatement select = SelectStatement.SELECT_ONE;
+ PhoenixStatement stmt = new PhoenixStatement(phoenixTable.pc);
+ ColumnResolver resolver = FromCompiler.getResolver(tableRef);
+ StatementContext context = new StatementContext(stmt, resolver, new Scan(), new SequenceManager(stmt));
+ Expression filterExpr = CalciteUtils.toExpression(filter, tmpImplementor);
+ filterExpr = WhereOptimizer.pushKeyExpressionsToScan(context, select, filterExpr);
+ scanRanges = context.getScanRanges();
+ if (!scanRanges.isPointLookup()
+ && !scanRanges.isDegenerate()
+ && !scanRanges.isEverything()) {
+ // TODO get the cf and timestamp right.
+ Scan scan = context.getScan();
+ byte[] cf = SchemaUtil.getEmptyColumnFamily(pTable);
+ statsHTable = phoenixTable.pc.getQueryServices()
+ .getTable(PhoenixDatabaseMetaData.SYSTEM_STATS_NAME_BYTES);
+ info = StatisticsUtil.readStatistics(
+ statsHTable, pTable.getPhysicalName().getBytes(),
+ new ImmutableBytesPtr(cf),
+ scan.getStartRow(),
+ scan.getStopRow(),
+ HConstants.LATEST_TIMESTAMP).getGuidePosts().get(cf);
+ }
+ } catch (SQLException | IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ if (statsHTable != null) {
+ try {
+ statsHTable.close();
+ } catch (IOException e) {
+ }
+ }
+ }
+ }
+ this.scanRanges = scanRanges;
+ this.filteredGuideposts = info;
+ }
+
+ private static ScanOrder getDefaultScanOrder(PhoenixTable table) {
+ //TODO why attribute value not correct in connectUsingModel??
+ //return table.pc.getQueryServices().getProps().getBoolean(
+ // QueryServices.FORCE_ROW_KEY_ORDER_ATTRIB,
+ // QueryServicesOptions.DEFAULT_FORCE_ROW_KEY_ORDER) ?
+ // ScanOrder.FORWARD : ScanOrder.NONE;
+ return ScanOrder.NONE;
+ }
+
+ private static List<RelCollation> reverse(List<RelCollation> collations) {
+ Builder<RelCollation> builder = ImmutableList.<RelCollation>builder();
+ for (RelCollation collation : collations) {
+ builder.add(CalciteUtils.reverseCollation(collation));
+ }
+ return builder.build();
+ }
+
+ public boolean isReverseScanEnabled() {
+ return table.unwrap(PhoenixTable.class).pc
+ .getQueryServices().getProps().getBoolean(
+ QueryServices.USE_REVERSE_SCAN_ATTRIB,
+ QueryServicesOptions.DEFAULT_USE_REVERSE_SCAN);
+ }
+
+ @Override
+ public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ assert inputs.isEmpty();
+ return this;
+ }
+
+ @Override
+ public RelWriter explainTerms(RelWriter pw) {
+ return super.explainTerms(pw)
+ .itemIf("filter", filter, filter != null)
+ .itemIf("scanOrder", scanOrder, scanOrder != ScanOrder.NONE);
+ }
+
+ @Override
+ public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ double byteCount;
+ PhoenixTable phoenixTable = table.unwrap(PhoenixTable.class);
+ if (scanRanges != null) {
+ if (scanRanges.isPointLookup()) {
+ byteCount = 1.0;
+ } else if (scanRanges.isDegenerate()) {
+ byteCount = 0.0;
+ } else if (scanRanges.isEverything()) {
+ byteCount = phoenixTable.byteCount;
+ } else {
+ if (filteredGuideposts != null) {
- byteCount = filteredGuideposts.getByteCount();
++ byteCount = 0;
++ for (long b : filteredGuideposts.getByteCounts()) {
++ byteCount += b;
++ }
+ // TODO why zero byteCount? a bug?
+ if (byteCount == 0 && filteredGuideposts.getGuidePostsCount() > 0) {
+ PTable pTable = phoenixTable.getTable();
+ byte[] emptyCf = SchemaUtil.getEmptyColumnFamily(pTable);
+ GuidePostsInfo info = pTable.getTableStats().getGuidePosts().get(emptyCf);
+ byteCount = phoenixTable.byteCount * filteredGuideposts.getGuidePostsCount() / info.getGuidePostsCount();
+ }
+ } else {
+ PTable pTable = phoenixTable.getTable();
+ byte[] emptyCf = SchemaUtil.getEmptyColumnFamily(pTable);
+ GuidePostsInfo info = pTable.getTableStats().getGuidePosts().get(emptyCf);
+ if (info != null) {
+ byteCount = phoenixTable.byteCount / info.getGuidePostsCount() / 2;
+ } else {
+ int pkCount = scanRanges.getBoundPkColumnCount();
+ byteCount = phoenixTable.byteCount * Math.pow(mq.getSelectivity(this, filter), pkCount);
+ }
+ }
+ }
+ } else {
+ byteCount = phoenixTable.byteCount;
+ }
+ byteCount *= rowCountFactor;
+ if (scanOrder != ScanOrder.NONE) {
+ // We don't want to make a big difference here. The idea is to avoid
+ // forcing row key order whenever the order is absolutely useless.
+ // E.g. in "select count(*) from t" we do not need the row key order;
+ // while in "select * from t order by pk0" we should force row key
+ // order to avoid sorting.
+ // Another case is "select pk0, count(*) from t", where we'd like to
+ // choose the row key ordered TableScan rel so that the Aggregate rel
+ // above it can be an stream aggregate, although at runtime this will
+ // eventually be an AggregatePlan, in which the "forceRowKeyOrder"
+ // flag takes no effect.
+ byteCount = addEpsilon(byteCount);
+ if (scanOrder == ScanOrder.REVERSE) {
+ byteCount = addEpsilon(byteCount);
+ }
+ }
+ return planner.getCostFactory().makeCost(0, byteCount, 0)
+ .multiplyBy(SERVER_FACTOR);
+ }
+
+ @Override
+ public double estimateRowCount(RelMetadataQuery mq) {
+ double rows = super.estimateRowCount(mq);
+ if (filter != null && !filter.isAlwaysTrue()) {
+ rows = rows * mq.getSelectivity(this, filter);
+ }
+
+ return rows * rowCountFactor;
+ }
+
+ @Override
+ public List<RelCollation> getCollationList() {
+ return getTraitSet().getTraits(RelCollationTraitDef.INSTANCE);
+ }
+
+ @Override
+ public QueryPlan implement(Implementor implementor) {
+ final PhoenixTable phoenixTable = table.unwrap(PhoenixTable.class);
+ PTable pTable = phoenixTable.getTable();
+ TableRef tableRef = new TableRef(CalciteUtils.createTempAlias(), pTable, HConstants.LATEST_TIMESTAMP, false);
+ implementor.setTableRef(tableRef);
+ try {
+ PhoenixStatement stmt = new PhoenixStatement(phoenixTable.pc);
+ ColumnResolver resolver = FromCompiler.getResolver(tableRef);
+ StatementContext context = new StatementContext(stmt, resolver, new Scan(), new SequenceManager(stmt));
+ SelectStatement select = SelectStatement.SELECT_ONE;
+ ImmutableIntList columnRefList = implementor.getCurrentContext().columnRefList;
+ Expression filterExpr = LiteralExpression.newConstant(Boolean.TRUE);
+ Expression dynamicFilter = null;
+ if (filter != null) {
+ ImmutableBitSet bitSet = InputFinder.analyze(filter).inputBitSet.addAll(columnRefList).build();
+ columnRefList = ImmutableIntList.copyOf(bitSet.asList());
+ filterExpr = CalciteUtils.toExpression(filter, implementor);
+ }
+ Expression rem = WhereOptimizer.pushKeyExpressionsToScan(context, select, filterExpr);
+ WhereCompiler.setScanFilter(context, select, rem, true, false);
+ // TODO This is not absolutely strict. We may have a filter like:
+ // pk = '0' and pk = $cor0 where $cor0 happens to get a sample value
+ // as '0', thus making the below test return false and adding an
+ // unnecessary dynamic filter. This would only be a performance bug though.
+ if (filter != null && !context.getScanRanges().equals(this.scanRanges)) {
+ dynamicFilter = filterExpr;
+ }
+ projectColumnFamilies(context.getScan(), phoenixTable.getTable(), columnRefList);
+ if (implementor.getCurrentContext().forceProject) {
+ TupleProjector tupleProjector = implementor.createTupleProjector();
+ TupleProjector.serializeProjectorIntoScan(context.getScan(), tupleProjector);
+ PTable projectedTable = implementor.createProjectedTable();
+ implementor.setTableRef(new TableRef(projectedTable));
+ }
+ Integer limit = null;
+ OrderBy orderBy = scanOrder == ScanOrder.NONE ?
+ OrderBy.EMPTY_ORDER_BY
+ : (scanOrder == ScanOrder.FORWARD ?
+ OrderBy.FWD_ROW_KEY_ORDER_BY
+ : OrderBy.REV_ROW_KEY_ORDER_BY);
+ ParallelIteratorFactory iteratorFactory = null;
+ return new ScanPlan(context, select, tableRef, RowProjector.EMPTY_PROJECTOR, limit, orderBy, iteratorFactory, true, dynamicFilter);
+ } catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void projectColumnFamilies(Scan scan, PTable table, ImmutableIntList columnRefList) {
+ scan.getFamilyMap().clear();
+ for (Integer index : columnRefList) {
+ PColumn column = table.getColumns().get(index);
+ PName familyName = column.getFamilyName();
+ if (familyName != null) {
+ scan.addFamily(familyName.getBytes());
+ }
+ }
+ }
+
+ private double addEpsilon(double d) {
+ assert d >= 0d;
+ final double d0 = d;
+ if (d < 10) {
+ // For small d, adding 1 would change the value significantly.
+ d *= 1.001d;
+ if (d != d0) {
+ return d;
+ }
+ }
+ // For medium d, add 1. Keeps integral values integral.
+ ++d;
+ if (d != d0) {
+ return d;
+ }
+ // For large d, adding 1 might not change the value. Add .1%.
+ // If d is NaN, this still will probably not change the value. That's OK.
+ d *= 1.001d;
+ return d;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
index 999c299,4be78a9..6909b23
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
@@@ -178,9 -178,8 +178,9 @@@ public class TupleProjectionCompiler
null, table.getTimeStamp(), table.getSequenceNumber(), table.getPKName(),
retainPKColumns ? table.getBucketNum() : null, projectedColumns, null,
null, Collections.<PTable>emptyList(), table.isImmutableRows(), Collections.<PName>emptyList(), null, null,
- table.isWALDisabled(), table.isMultiTenant(), table.getStoreNulls(), table.getViewType(), table.getViewIndexId(),
- null, table.rowKeyOrderOptimizable(), table.isTransactional(), table.getUpdateCacheFrequency(), table.getIndexDisableTimestamp());
+ table.isWALDisabled(), retainPKColumns ? table.isMultiTenant() : false, table.getStoreNulls(), table.getViewType(),
+ retainPKColumns ? table.getViewIndexId() : null, null, table.rowKeyOrderOptimizable(), table.isTransactional(),
- table.getUpdateCacheFrequency());
++ table.getUpdateCacheFrequency(), table.getIndexDisableTimestamp());
}
// For extracting column references from single select statement
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
index e38e495,d51e6c8..673a641
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
@@@ -87,10 -83,10 +88,10 @@@ public class ScanPlan extends BaseQuery
this(context, statement, table, projector, limit, orderBy, parallelIteratorFactory, allowPageFilter, null);
}
- private ScanPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector, Integer limit, OrderBy orderBy, ParallelIteratorFactory parallelIteratorFactory, boolean allowPageFilter, Expression dynamicFilter) throws SQLException {
+ public ScanPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector, Integer limit, OrderBy orderBy, ParallelIteratorFactory parallelIteratorFactory, boolean allowPageFilter, Expression dynamicFilter) throws SQLException {
super(context, statement, table, projector, context.getBindManager().getParameterMetaData(), limit, orderBy, GroupBy.EMPTY_GROUP_BY,
parallelIteratorFactory != null ? parallelIteratorFactory :
- buildResultIteratorFactory(context, table, orderBy, limit, allowPageFilter), dynamicFilter);
+ buildResultIteratorFactory(context, statement, table, orderBy, limit, allowPageFilter), dynamicFilter);
this.allowPageFilter = allowPageFilter;
if (!orderBy.getOrderByExpressions().isEmpty()) { // TopN
int thresholdBytes = context.getConnection().getQueryServices().getProps().getInt(
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/stats/StatisticsWriter.java
index 830b1d6,f8c888d..6da829b
--- 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
@@@ -217,9 -217,9 +217,9 @@@ public class StatisticsWriter implement
public void deleteStats(Region region, StatisticsCollector tracker, ImmutableBytesPtr fam, List<Mutation> mutations)
throws IOException {
- long timeStamp = clientTimeStamp == StatisticsCollector.NO_TIMESTAMP ? tracker.getMaxTimeStamp()
+ long timeStamp = clientTimeStamp == DefaultStatisticsCollector.NO_TIMESTAMP ? tracker.getMaxTimeStamp()
: clientTimeStamp;
- List<Result> statsForRegion = StatisticsUtil.readStatistics(statsWriterTable, tableName, fam,
+ List<Result> statsForRegion = StatisticsUtil.readStatisticsForDelete(statsWriterTable, tableName, fam,
region.getRegionInfo().getStartKey(), region.getRegionInfo().getEndKey(), timeStamp);
for (Result result : statsForRegion) {
mutations.add(new Delete(result.getRow(), timeStamp - 1));
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/58ec2579/pom.xml
----------------------------------------------------------------------
diff --cc pom.xml
index c3fb0f4,b27c3b9..84e0dbf
--- a/pom.xml
+++ b/pom.xml
@@@ -107,11 -107,12 +107,12 @@@
<commons-codec.version>1.7</commons-codec.version>
<htrace.version>3.1.0-incubating</htrace.version>
<collections.version>3.2.1</collections.version>
- <jodatime.version>2.7</jodatime.version>
+ <!-- Do not change jodatime.version until HBASE-15199 is fixed -->
+ <jodatime.version>1.6</jodatime.version>
<joni.version>2.1.2</joni.version>
- <calcite.version>1.6.0</calcite.version>
+ <calcite.version>1.7.0-SNAPSHOT</calcite.version>
<jettyVersion>8.1.7.v20120910</jettyVersion>
- <tephra.version>0.6.4</tephra.version>
+ <tephra.version>0.7.0</tephra.version>
<spark.version>1.5.2</spark.version>
<scala.version>2.10.4</scala.version>
<scala.binary.version>2.10</scala.binary.version>