You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sa...@apache.org on 2016/09/19 23:46:23 UTC
[9/9] phoenix git commit: PHOENIX-2565 Store data for immutable
tables in single KeyValue (Thomas DSilva)
PHOENIX-2565 Store data for immutable tables in single KeyValue (Thomas DSilva)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/920be8fe
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/920be8fe
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/920be8fe
Branch: refs/heads/encodecolumns
Commit: 920be8fecfcbf2170ba040b3b52d1fcd0a1622fd
Parents: cc472a8
Author: Samarth <sa...@salesforce.com>
Authored: Mon Sep 19 16:45:50 2016 -0700
Committer: Samarth <sa...@salesforce.com>
Committed: Mon Sep 19 16:45:50 2016 -0700
----------------------------------------------------------------------
.../apache/phoenix/end2end/UpsertValuesIT.java | 5 +-
.../end2end/index/IndexExpressionIT.java | 7 +-
.../apache/phoenix/end2end/index/IndexIT.java | 24 ++--
.../phoenix/compile/ExpressionCompiler.java | 3 +-
.../apache/phoenix/compile/JoinCompiler.java | 2 +-
.../apache/phoenix/compile/QueryCompiler.java | 2 +-
.../compile/TupleProjectionCompiler.java | 2 +-
.../apache/phoenix/compile/WhereCompiler.java | 31 +++++-
.../coprocessor/BaseScannerRegionObserver.java | 1 +
.../coprocessor/MetaDataEndpointImpl.java | 1 +
.../UngroupedAggregateRegionObserver.java | 2 +-
.../apache/phoenix/execute/BaseQueryPlan.java | 13 ++-
.../apache/phoenix/execute/MutationState.java | 10 +-
.../expression/ArrayColumnExpression.java | 109 +++++++++++++++++++
.../expression/ArrayConstructorExpression.java | 1 +
.../phoenix/expression/ExpressionType.java | 3 +-
.../phoenix/expression/LiteralExpression.java | 11 +-
.../visitor/CloneExpressionVisitor.java | 6 +
.../expression/visitor/ExpressionVisitor.java | 2 +
...lumnWithKeyValueColumnExpressionVisitor.java | 37 +++++++
.../StatelessTraverseAllExpressionVisitor.java | 7 +-
.../StatelessTraverseNoExpressionVisitor.java | 7 +-
.../filter/MultiKeyValueComparisonFilter.java | 6 +
.../SingleCQKeyValueComparisonFilter.java | 3 +-
.../filter/SingleKeyValueComparisonFilter.java | 8 ++
.../apache/phoenix/hbase/index/ValueGetter.java | 1 +
.../apache/phoenix/index/IndexMaintainer.java | 101 ++++++++++++++---
.../apache/phoenix/index/PhoenixIndexCodec.java | 2 +-
.../phoenix/iterate/BaseResultIterators.java | 7 +-
.../query/ConnectionlessQueryServicesImpl.java | 1 -
.../org/apache/phoenix/schema/ColumnRef.java | 7 +-
.../apache/phoenix/schema/MetaDataClient.java | 16 ++-
.../java/org/apache/phoenix/schema/PTable.java | 3 +-
.../org/apache/phoenix/schema/PTableImpl.java | 77 ++++++++++---
.../apache/phoenix/schema/TableProperty.java | 2 +-
.../apache/phoenix/schema/tuple/BaseTuple.java | 30 +++++
.../apache/phoenix/util/EncodedColumnsUtil.java | 19 +++-
.../java/org/apache/phoenix/util/IndexUtil.java | 52 +++++----
.../java/org/apache/phoenix/util/ScanUtil.java | 4 +-
.../org/apache/phoenix/util/SchemaUtil.java | 18 +++
.../phoenix/index/IndexMaintainerTest.java | 2 +-
41 files changed, 547 insertions(+), 98 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java
index 28574ed..24ca24a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertValuesIT.java
@@ -35,8 +35,11 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Time;
import java.sql.Timestamp;
+import java.util.Map.Entry;
+import java.util.NavigableMap;
import java.util.Properties;
+import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
@@ -839,7 +842,7 @@ public class UpsertValuesIT extends BaseClientManagedTimeIT {
assertEquals("KV2", rs.getString(2));
assertFalse(rs.next());
- // Verify now that the data was correctly added to the mutable index too.
+ // Verify now that the data was correctly added to the immutable index too.
stmt = conn.prepareStatement("SELECT KV2 FROM " + tableName + " WHERE PK2 = ? AND KV1 = ?");
stmt.setDate(1, upsertedDate);
stmt.setString(2, "KV1");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexExpressionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexExpressionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexExpressionIT.java
index 21da43a..bdb5b93 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexExpressionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexExpressionIT.java
@@ -30,9 +30,14 @@ import java.sql.SQLException;
import java.util.Properties;
import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
import org.apache.phoenix.end2end.BaseHBaseManagedTimeIT;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.execute.CommitException;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.util.DateUtil;
import org.apache.phoenix.util.IndexUtil;
@@ -1300,7 +1305,7 @@ public class IndexExpressionIT extends BaseHBaseManagedTimeIT {
try {
conn.createStatement().execute(
"CREATE TABLE t (k VARCHAR NOT NULL PRIMARY KEY, v VARCHAR) "
- + (mutable ? "IMMUTABLE_ROWS=true" : ""));
+ + (!mutable ? "IMMUTABLE_ROWS=true" : ""));
String query = "SELECT * FROM t";
ResultSet rs = conn.createStatement().executeQuery(query);
assertFalse(rs.next());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexIT.java
index b7537a6..d1bb085 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/IndexIT.java
@@ -770,23 +770,23 @@ public class IndexIT extends BaseHBaseManagedTimeIT {
stmt.execute();
conn.commit();
- // make sure the index is working as expected
- query = "SELECT * FROM " + fullIndexName;
+ query = "SELECT /*+ NO_INDEX */ * FROM " + testTable;
rs = conn.createStatement().executeQuery(query);
assertTrue(rs.next());
- assertEquals("x", rs.getString(1));
- assertEquals("1", rs.getString(2));
- assertEquals("a", rs.getString(3));
+ assertEquals("a", rs.getString(1));
+ assertEquals("x", rs.getString(2));
+ assertEquals("1", rs.getString(3));
assertTrue(rs.next());
- assertEquals("y", rs.getString(1));
- assertEquals("2", rs.getString(2));
- assertEquals("b", rs.getString(3));
+ assertEquals("b", rs.getString(1));
+ assertEquals("y", rs.getString(2));
+ assertEquals("2", rs.getString(3));
assertTrue(rs.next());
- assertEquals("z", rs.getString(1));
- assertEquals("3", rs.getString(2));
- assertEquals("c", rs.getString(3));
+ assertEquals("c", rs.getString(1));
+ assertEquals("z", rs.getString(2));
+ assertEquals("3", rs.getString(3));
assertFalse(rs.next());
+ // make sure the index is working as expected
query = "SELECT * FROM " + testTable;
rs = conn.createStatement().executeQuery("EXPLAIN " + query);
if (localIndex) {
@@ -855,7 +855,7 @@ public class IndexIT extends BaseHBaseManagedTimeIT {
} else {
assertEquals("CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + fullIndexName + " ['1']", QueryUtil.getExplainPlan(rs));
}
-
+
rs = conn.createStatement().executeQuery(query);
assertTrue(rs.next());
assertEquals("a",rs.getString(1));
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java
index 1623cab..cfd0afc 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ExpressionCompiler.java
@@ -116,6 +116,7 @@ import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PDatum;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeyValueAccessor;
import org.apache.phoenix.schema.SortOrder;
@@ -408,7 +409,7 @@ public class ExpressionCompiler extends UnsupportedAllParseNodeVisitor<Expressio
return LiteralExpression.newConstant(column.getDataType().toObject(ptr), column.getDataType());
}
if (tableRef.equals(context.getCurrentTable()) && !SchemaUtil.isPKColumn(column)) { // project only kv columns
- context.getScan().addColumn(column.getFamilyName().getBytes(), EncodedColumnsUtil.getColumnQualifier(column, tableRef.getTable()));
+ EncodedColumnsUtil.setColumns(column, tableRef.getTable(), context.getScan());
}
Expression expression = ref.newColumnExpression(node.isTableNameCaseSensitive(), node.isCaseSensitive());
Expression wrappedExpression = wrapGroupByExpression(expression);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
index 36c93f7..489b993 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
@@ -716,7 +716,7 @@ public class JoinCompiler {
if (columnRef.getTableRef().equals(tableRef)
&& !SchemaUtil.isPKColumn(columnRef.getColumn())
&& !(columnRef instanceof LocalIndexColumnRef)) {
- scan.addColumn(columnRef.getColumn().getFamilyName().getBytes(), EncodedColumnsUtil.getColumnQualifier(columnRef.getColumn(), tableRef.getTable()));
+ EncodedColumnsUtil.setColumns(columnRef.getColumn(), tableRef.getTable(), scan);
}
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java
index 2258f28..5126c8b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java
@@ -94,7 +94,7 @@ public class QueryCompiler {
*/
private static final String LOAD_COLUMN_FAMILIES_ON_DEMAND_ATTR = "_ondemand_";
private final PhoenixStatement statement;
- private final Scan scan;
+ private final Scan scan;
private final Scan originalScan;
private final ColumnResolver resolver;
private final SelectStatement select;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
index b6bb5da..3898969 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/TupleProjectionCompiler.java
@@ -141,7 +141,7 @@ public class TupleProjectionCompiler {
projectedColumns.add(column);
// Wildcard or FamilyWildcard will be handled by ProjectionCompiler.
if (!isWildcard && !families.contains(sourceColumn.getFamilyName())) {
- context.getScan().addColumn(sourceColumn.getFamilyName().getBytes(), EncodedColumnsUtil.getColumnQualifier(column, table));
+ EncodedColumnsUtil.setColumns(column, table, context.getScan());
}
}
// add LocalIndexDataColumnRef
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/compile/WhereCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/WhereCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/WhereCompiler.java
index 5e64209..63ad9c0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/WhereCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/WhereCompiler.java
@@ -29,6 +29,8 @@ import org.apache.hadoop.hbase.filter.Filter;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.expression.AndExpression;
+import org.apache.phoenix.expression.ArrayColumnExpression;
+import org.apache.phoenix.expression.ColumnExpression;
import org.apache.phoenix.expression.Determinism;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
@@ -51,6 +53,7 @@ import org.apache.phoenix.schema.ColumnNotFoundException;
import org.apache.phoenix.schema.ColumnRef;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.TypeMismatchException;
@@ -169,12 +172,14 @@ public class WhereCompiler {
public Expression visit(ColumnParseNode node) throws SQLException {
ColumnRef ref = resolveColumn(node);
TableRef tableRef = ref.getTableRef();
+ ColumnExpression newColumnExpression = ref.newColumnExpression(node.isTableNameCaseSensitive(), node.isCaseSensitive());
if (tableRef.equals(context.getCurrentTable()) && !SchemaUtil.isPKColumn(ref.getColumn())) {
+ byte[] cq = tableRef.getTable().getStorageScheme() == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL
+ ? ref.getColumn().getFamilyName().getBytes() : EncodedColumnsUtil.getColumnQualifier(ref.getColumn(), tableRef.getTable());
// track the where condition columns. Later we need to ensure the Scan in HRS scans these column CFs
- byte[] cq = EncodedColumnsUtil.getColumnQualifier(ref.getColumn(), tableRef.getTable());
context.addWhereCoditionColumn(ref.getColumn().getFamilyName().getBytes(), cq);
}
- return ref.newColumnExpression(node.isTableNameCaseSensitive(), node.isCaseSensitive());
+ return newColumnExpression;
}
@Override
@@ -223,6 +228,22 @@ public class WhereCompiler {
}
}
+
+ public void increment(ArrayColumnExpression column) {
+ switch (count) {
+ case NONE:
+ count = Count.SINGLE;
+ this.column = column.getArrayExpression();
+ break;
+ case SINGLE:
+ count = column.getArrayExpression().equals(this.column) ? Count.SINGLE : Count.MULTIPLE;
+ break;
+ case MULTIPLE:
+ break;
+
+ }
+ }
+
public Count getCount() {
return count;
}
@@ -257,6 +278,12 @@ public class WhereCompiler {
counter.increment(expression);
return null;
}
+
+ @Override
+ public Void visit(ArrayColumnExpression expression) {
+ counter.increment(expression);
+ return null;
+ }
});
switch (counter.getCount()) {
case NONE:
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
index 5922b5d..ee65575 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/BaseScannerRegionObserver.java
@@ -86,6 +86,7 @@ abstract public class BaseScannerRegionObserver extends BaseRegionObserver {
public static final String LOCAL_INDEX_BUILD = "_LocalIndexBuild";
public static final String LOCAL_INDEX_JOIN_SCHEMA = "_LocalIndexJoinSchema";
public static final String DATA_TABLE_COLUMNS_TO_JOIN = "_DataTableColumnsToJoin";
+ public static final String COLUMNS_STORED_IN_SINGLE_CELL = "_ColumnsStoredInSingleCell";
public static final String VIEW_CONSTANTS = "_ViewConstants";
public static final String EXPECTED_UPPER_REGION_KEY = "_ExpectedUpperRegionKey";
public static final String REVERSE_SCAN = "_ReverseScan";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index de68551..d8372ce 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -944,6 +944,7 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
StorageScheme storageScheme = storageSchemeKv == null ? StorageScheme.NON_ENCODED_COLUMN_NAMES : StorageScheme
.fromSerializedValue((byte)PTinyint.INSTANCE.toObject(storageSchemeKv.getValueArray(),
storageSchemeKv.getValueOffset(), storageSchemeKv.getValueLength()));
+
List<PColumn> columns = Lists.newArrayListWithExpectedSize(columnCount);
List<PTable> indexes = new ArrayList<PTable>();
List<PName> physicalTables = new ArrayList<PName>();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
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 c1ef0b3..3129ef8 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
@@ -498,7 +498,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
Put put = maintainer.buildUpdateMutation(kvBuilder,
valueGetter, ptr, results.get(0).getTimestamp(),
env.getRegion().getRegionInfo().getStartKey(),
- env.getRegion().getRegionInfo().getEndKey());
+ env.getRegion().getRegionInfo().getEndKey(), false);
indexMutations.add(put);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
index 3453120..0dceaea 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BaseQueryPlan.java
@@ -64,6 +64,7 @@ import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.trace.TracingIterator;
@@ -298,6 +299,7 @@ public abstract class BaseQueryPlan implements QueryPlan {
// TODO: can have an hint to skip joining back to data table, in that case if any column to
// project is not present in the index then we need to skip this plan.
if (!dataColumns.isEmpty()) {
+ // Set data columns to be join back from data table.
PTable parentTable = context.getCurrentTable().getTable();
String parentSchemaName = parentTable.getParentSchemaName().getString();
String parentTableName = parentTable.getParentTableName().getString();
@@ -420,10 +422,17 @@ public abstract class BaseQueryPlan implements QueryPlan {
ByteArrayOutputStream stream = new ByteArrayOutputStream();
try {
DataOutputStream output = new DataOutputStream(stream);
+ boolean storeColsInSingleCell = dataTable.getStorageScheme() == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL;
+ if (storeColsInSingleCell) {
+ // if storeColsInSingleCell is true all columns of a given column family are stored in a single cell
+ scan.setAttribute(BaseScannerRegionObserver.COLUMNS_STORED_IN_SINGLE_CELL, QueryConstants.EMPTY_COLUMN_VALUE_BYTES);
+ }
WritableUtils.writeVInt(output, dataColumns.size());
for (PColumn column : dataColumns) {
- Bytes.writeByteArray(output, column.getFamilyName().getBytes());
- Bytes.writeByteArray(output, EncodedColumnsUtil.getColumnQualifier(column, dataTable));
+ byte[] cf = column.getFamilyName().getBytes();
+ byte[] cq = EncodedColumnsUtil.getColumnQualifier(column, dataTable);
+ Bytes.writeByteArray(output, cf);
+ Bytes.writeByteArray(output, cq);
}
scan.setAttribute(BaseScannerRegionObserver.DATA_TABLE_COLUMNS_TO_JOIN, stream.toByteArray());
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index 51ca59c..34dca46 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -582,7 +582,7 @@ public class MutationState implements SQLCloseable {
List<Mutation> indexMutations;
try {
indexMutations =
- IndexUtil.generateIndexData(table, index, mutationsPertainingToIndex,
+ IndexUtil.generateIndexData(table, index, values, mutationsPertainingToIndex,
connection.getKeyValueBuilder(), connection);
// we may also have to include delete mutations for immutable tables if we are not processing all the tables in the mutations map
if (!sendAll) {
@@ -616,6 +616,7 @@ public class MutationState implements SQLCloseable {
Iterator<Map.Entry<ImmutableBytesPtr, RowMutationState>> iterator =
values.entrySet().iterator();
long timestampToUse = timestamp;
+ Map<ImmutableBytesPtr, RowMutationState> modifiedValues = Maps.newHashMap();
while (iterator.hasNext()) {
Map.Entry<ImmutableBytesPtr, RowMutationState> rowEntry = iterator.next();
ImmutableBytesPtr key = rowEntry.getKey();
@@ -623,6 +624,10 @@ public class MutationState implements SQLCloseable {
if (tableWithRowTimestampCol) {
RowTimestampColInfo rowTsColInfo = state.getRowTimestampColInfo();
if (rowTsColInfo.useServerTimestamp()) {
+ // since we are about to modify the byte[] stored in key (which changes its hashcode)
+ // we need to remove the entry from the values map and add a new entry with the modified byte[]
+ modifiedValues.put(key, state);
+ iterator.remove();
// regenerate the key with this timestamp.
key = getNewRowKeyWithRowTimestamp(key, timestampToUse, table);
} else {
@@ -654,6 +659,7 @@ public class MutationState implements SQLCloseable {
if (mutationsPertainingToIndex != null) mutationsPertainingToIndex
.addAll(rowMutationsPertainingToIndex);
}
+ values.putAll(modifiedValues);
}
/**
@@ -1383,7 +1389,7 @@ public class MutationState implements SQLCloseable {
this.rowTsColInfo = rowTsColInfo;
}
- Map<PColumn, byte[]> getColumnValues() {
+ public Map<PColumn, byte[]> getColumnValues() {
return columnValues;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayColumnExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayColumnExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayColumnExpression.java
new file mode 100644
index 0000000..aaa9f75
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayColumnExpression.java
@@ -0,0 +1,109 @@
+/*
+ * 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.expression;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.phoenix.expression.visitor.ExpressionVisitor;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PDatum;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.schema.types.PVarchar;
+
+/**
+ *
+ * Class to access a column that is stored in a KeyValue that contains all
+ * columns for a given column family (stored in an array)
+ *
+ */
+public class ArrayColumnExpression extends ColumnExpression {
+
+ private String displayName; // client-side only
+ private int index;
+ // expression that represents the array (where all cols are stored in a single key value)
+ private KeyValueColumnExpression arrayExpression;
+ // expression that represents this column if (it were stored as a regular key value)
+ private KeyValueColumnExpression origKVExpression;
+
+ public ArrayColumnExpression() {
+ }
+
+ public ArrayColumnExpression(PDatum column, byte[] cf, int index) {
+ super(column);
+ this.index = index;
+ this.arrayExpression = new KeyValueColumnExpression(column, cf, cf);
+ }
+
+ public ArrayColumnExpression(PColumn column, String displayName, boolean encodedColumnName) {
+ super(column);
+ // array indexes are 1-based
+ this.index = column.getEncodedColumnQualifier()+1;
+ byte[] cf = column.getFamilyName().getBytes();
+ this.arrayExpression = new KeyValueColumnExpression(column, cf, cf);
+ this.origKVExpression = new KeyValueColumnExpression(column, displayName, encodedColumnName);
+ this.displayName = displayName;
+ }
+
+ @Override
+ public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
+ return PArrayDataType.positionAtArrayElement(tuple, ptr, index, arrayExpression, PVarbinary.INSTANCE, null);
+ }
+
+ @Override
+ public <T> T accept(ExpressionVisitor<T> visitor) {
+ return visitor.visit(this);
+ }
+
+ @Override
+ public void readFields(DataInput input) throws IOException {
+ super.readFields(input);
+ index = WritableUtils.readVInt(input);
+ arrayExpression = new KeyValueColumnExpression();
+ arrayExpression.readFields(input);
+ origKVExpression = new KeyValueColumnExpression();
+ origKVExpression.readFields(input);
+ }
+
+ @Override
+ public void write(DataOutput output) throws IOException {
+ super.write(output);
+ WritableUtils.writeVInt(output, index);
+ arrayExpression.write(output);
+ origKVExpression.write(output);
+ }
+
+ public KeyValueColumnExpression getArrayExpression() {
+ return arrayExpression;
+ }
+
+ public KeyValueColumnExpression getKeyValueExpression() {
+ return origKVExpression;
+ }
+
+ @Override
+ public String toString() {
+ return displayName;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayConstructorExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayConstructorExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayConstructorExpression.java
index d8df29a..f513664 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayConstructorExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/ArrayConstructorExpression.java
@@ -115,6 +115,7 @@ public class ArrayConstructorExpression extends BaseCompoundExpression {
offsetPos[i] = byteStream.size();
oStream.write(ptr.get(), ptr.getOffset(), ptr.getLength());
oStream.write(PArrayDataType.getSeparatorByte(rowKeyOrderOptimizable, getSortOrder()));
+ nNulls = 0;
}
} else { // No nulls for fixed length
oStream.write(ptr.get(), ptr.getOffset(), ptr.getLength());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
index e750732..64f0ffa 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
@@ -280,7 +280,8 @@ public enum ExpressionType {
FloorYearExpression(FloorYearExpression.class),
CeilWeekExpression(CeilWeekExpression.class),
CeilMonthExpression(CeilMonthExpression.class),
- CeilYearExpression(CeilYearExpression.class);
+ CeilYearExpression(CeilYearExpression.class),
+ ArrayColumnExpression(ArrayColumnExpression.class);
ExpressionType(Class<? extends Expression> clazz) {
this.clazz = clazz;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
index 90882a2..f20d7e2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/LiteralExpression.java
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.sql.SQLException;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.expression.visitor.ExpressionVisitor;
import org.apache.phoenix.schema.IllegalDataException;
@@ -214,6 +215,11 @@ public class LiteralExpression extends BaseTerminalExpression {
public LiteralExpression() {
}
+
+ public LiteralExpression(byte[] byteValue) {
+ this.byteValue = byteValue!=null ? byteValue : ByteUtil.EMPTY_BYTE_ARRAY;
+ this.determinism = Determinism.ALWAYS;
+ }
private LiteralExpression(PDataType type, Determinism determinism) {
this(null, type, ByteUtil.EMPTY_BYTE_ARRAY, determinism);
@@ -242,7 +248,10 @@ public class LiteralExpression extends BaseTerminalExpression {
@Override
public String toString() {
- if (value == null) {
+ if (value == null && byteValue!=null) {
+ return Bytes.toStringBinary(byteValue);
+ }
+ else if (value == null) {
return "null";
}
// TODO: move into PDataType?
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/CloneExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/CloneExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/CloneExpressionVisitor.java
index 00ece40..15a9f74 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/CloneExpressionVisitor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/CloneExpressionVisitor.java
@@ -26,6 +26,7 @@ import org.apache.phoenix.expression.ArrayConstructorExpression;
import org.apache.phoenix.expression.CaseExpression;
import org.apache.phoenix.expression.CoerceExpression;
import org.apache.phoenix.expression.ComparisonExpression;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.CorrelateVariableFieldAccessExpression;
import org.apache.phoenix.expression.DivideExpression;
import org.apache.phoenix.expression.Expression;
@@ -80,6 +81,11 @@ public abstract class CloneExpressionVisitor extends TraverseAllExpressionVisito
public Expression visit(KeyValueColumnExpression node) {
return node;
}
+
+ @Override
+ public Expression visit(ArrayColumnExpression node) {
+ return node;
+ }
@Override
public Expression visit(ProjectedColumnExpression node) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ExpressionVisitor.java
index 31f340d..100f099 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ExpressionVisitor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ExpressionVisitor.java
@@ -27,6 +27,7 @@ import org.apache.phoenix.expression.ArrayConstructorExpression;
import org.apache.phoenix.expression.CaseExpression;
import org.apache.phoenix.expression.CoerceExpression;
import org.apache.phoenix.expression.ComparisonExpression;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.CorrelateVariableFieldAccessExpression;
import org.apache.phoenix.expression.DivideExpression;
import org.apache.phoenix.expression.Expression;
@@ -113,6 +114,7 @@ public interface ExpressionVisitor<E> {
public E visit(LiteralExpression node);
public E visit(RowKeyColumnExpression node);
public E visit(KeyValueColumnExpression node);
+ public E visit(ArrayColumnExpression node);
public E visit(ProjectedColumnExpression node);
public E visit(SequenceValueExpression node);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ReplaceArrayColumnWithKeyValueColumnExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ReplaceArrayColumnWithKeyValueColumnExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ReplaceArrayColumnWithKeyValueColumnExpressionVisitor.java
new file mode 100644
index 0000000..7ca6d9e
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/ReplaceArrayColumnWithKeyValueColumnExpressionVisitor.java
@@ -0,0 +1,37 @@
+/*
+ * 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.expression.visitor;
+
+import java.util.List;
+
+import org.apache.phoenix.expression.ArrayColumnExpression;
+import org.apache.phoenix.expression.Expression;
+
+public class ReplaceArrayColumnWithKeyValueColumnExpressionVisitor extends CloneExpressionVisitor {
+
+ @Override
+ public boolean isCloneNode(Expression node, List<Expression> children) {
+ return !children.equals(node.getChildren());
+ }
+
+ @Override
+ public Expression visit(ArrayColumnExpression node) {
+ return node.getKeyValueExpression();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseAllExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseAllExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseAllExpressionVisitor.java
index 3b7067a..9e50bc4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseAllExpressionVisitor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseAllExpressionVisitor.java
@@ -26,9 +26,9 @@ import org.apache.phoenix.expression.ArrayConstructorExpression;
import org.apache.phoenix.expression.CaseExpression;
import org.apache.phoenix.expression.CoerceExpression;
import org.apache.phoenix.expression.ComparisonExpression;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.CorrelateVariableFieldAccessExpression;
import org.apache.phoenix.expression.DivideExpression;
-import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.InListExpression;
import org.apache.phoenix.expression.IsNullExpression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
@@ -121,6 +121,11 @@ public class StatelessTraverseAllExpressionVisitor<E> extends TraverseAllExpress
}
@Override
+ public E visit(ArrayColumnExpression node) {
+ return null;
+ }
+
+ @Override
public E visit(ProjectedColumnExpression node) {
return null;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseNoExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseNoExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseNoExpressionVisitor.java
index 83b28bd..1a2f2cc 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseNoExpressionVisitor.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/StatelessTraverseNoExpressionVisitor.java
@@ -26,9 +26,9 @@ import org.apache.phoenix.expression.ArrayConstructorExpression;
import org.apache.phoenix.expression.CaseExpression;
import org.apache.phoenix.expression.CoerceExpression;
import org.apache.phoenix.expression.ComparisonExpression;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.CorrelateVariableFieldAccessExpression;
import org.apache.phoenix.expression.DivideExpression;
-import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.InListExpression;
import org.apache.phoenix.expression.IsNullExpression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
@@ -114,6 +114,11 @@ public class StatelessTraverseNoExpressionVisitor<E> extends TraverseNoExpressio
public E visit(RowKeyColumnExpression node) {
return null;
}
+
+ @Override
+ public E visit(ArrayColumnExpression node) {
+ return null;
+ }
@Override
public E visit(KeyValueColumnExpression node) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/filter/MultiKeyValueComparisonFilter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/filter/MultiKeyValueComparisonFilter.java b/phoenix-core/src/main/java/org/apache/phoenix/filter/MultiKeyValueComparisonFilter.java
index c3d52a4..5909286 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/filter/MultiKeyValueComparisonFilter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/filter/MultiKeyValueComparisonFilter.java
@@ -26,6 +26,7 @@ import java.util.TreeSet;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
import org.apache.phoenix.expression.visitor.ExpressionVisitor;
@@ -187,6 +188,11 @@ public abstract class MultiKeyValueComparisonFilter extends BooleanExpressionFil
inputTuple.addColumn(expression.getColumnFamily(), expression.getColumnQualifier());
return null;
}
+ @Override
+ public Void visit(ArrayColumnExpression expression) {
+ inputTuple.addColumn(expression.getArrayExpression().getColumnFamily(), expression.getArrayExpression().getColumnQualifier());
+ return null;
+ }
};
expression.accept(visitor);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleCQKeyValueComparisonFilter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleCQKeyValueComparisonFilter.java b/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleCQKeyValueComparisonFilter.java
index 0d904bc..195c89c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleCQKeyValueComparisonFilter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleCQKeyValueComparisonFilter.java
@@ -47,7 +47,8 @@ public class SingleCQKeyValueComparisonFilter extends SingleKeyValueComparisonFi
public static SingleCQKeyValueComparisonFilter parseFrom(final byte [] pbBytes) throws DeserializationException {
try {
- return (SingleCQKeyValueComparisonFilter)Writables.getWritable(pbBytes, new SingleCQKeyValueComparisonFilter());
+ SingleCQKeyValueComparisonFilter writable = (SingleCQKeyValueComparisonFilter)Writables.getWritable(pbBytes, new SingleCQKeyValueComparisonFilter());
+ return writable;
} catch (IOException e) {
throw new DeserializationException(e);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleKeyValueComparisonFilter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleKeyValueComparisonFilter.java b/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleKeyValueComparisonFilter.java
index 07f7072..527b948 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleKeyValueComparisonFilter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/filter/SingleKeyValueComparisonFilter.java
@@ -22,11 +22,13 @@ import java.io.IOException;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
import org.apache.phoenix.expression.visitor.StatelessTraverseAllExpressionVisitor;
import org.apache.phoenix.expression.visitor.TraverseAllExpressionVisitor;
import org.apache.phoenix.schema.tuple.SingleKeyValueTuple;
+import org.apache.phoenix.schema.tuple.Tuple;
@@ -61,6 +63,12 @@ public abstract class SingleKeyValueComparisonFilter extends BooleanExpressionFi
cq = expression.getColumnQualifier();
return null;
}
+ @Override
+ public Void visit(ArrayColumnExpression expression) {
+ cf = expression.getArrayExpression().getColumnFamily();
+ cq = expression.getArrayExpression().getColumnQualifier();
+ return null;
+ }
};
expression.accept(visitor);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/ValueGetter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/ValueGetter.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/ValueGetter.java
index bcadc2b..19797cf 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/ValueGetter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/ValueGetter.java
@@ -35,4 +35,5 @@ public interface ValueGetter {
public ImmutableBytesWritable getLatestValue(ColumnReference ref) throws IOException;
public byte[] getRowKey();
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
index 2ad0c8d..a3f2a1b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/IndexMaintainer.java
@@ -35,7 +35,6 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
-
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HConstants;
@@ -52,10 +51,14 @@ import org.apache.phoenix.compile.ColumnResolver;
import org.apache.phoenix.compile.FromCompiler;
import org.apache.phoenix.compile.IndexExpressionCompiler;
import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.expression.ArrayColumnExpression;
+import org.apache.phoenix.expression.ArrayConstructorExpression;
import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.expression.ExpressionType;
import org.apache.phoenix.expression.KeyValueColumnExpression;
+import org.apache.phoenix.expression.LiteralExpression;
import org.apache.phoenix.expression.visitor.KeyValueExpressionVisitor;
+import org.apache.phoenix.expression.visitor.ReplaceArrayColumnWithKeyValueColumnExpressionVisitor;
import org.apache.phoenix.hbase.index.ValueGetter;
import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
@@ -76,6 +79,7 @@ import org.apache.phoenix.schema.PIndexState;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.RowKeySchema;
import org.apache.phoenix.schema.SaltingUtil;
@@ -83,8 +87,11 @@ import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.ValueSchema;
import org.apache.phoenix.schema.ValueSchema.Field;
+import org.apache.phoenix.schema.tuple.BaseTuple;
import org.apache.phoenix.schema.tuple.ValueGetterTuple;
import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.schema.types.PVarchar;
import org.apache.phoenix.util.BitSet;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.EncodedColumnsUtil;
@@ -295,6 +302,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
private boolean indexWALDisabled;
private boolean isLocalIndex;
private boolean immutableRows;
+ private boolean storeColsInSingleCell;
// Transient state
private final boolean isDataTableSalted;
@@ -386,6 +394,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
// TODO: check whether index is immutable or not. Currently it's always false so checking
// data table is with immutable rows or not.
this.immutableRows = dataTable.isImmutableRows();
+ this.storeColsInSingleCell = index.getStorageScheme() == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL;
int indexColByteSize = 0;
ColumnResolver resolver = null;
List<ParseNode> parseNodes = new ArrayList<ParseNode>(1);
@@ -455,7 +464,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
initCachedState();
}
- public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey, byte[] regionEndKey) {
+ public byte[] buildRowKey(ValueGetter valueGetter, ImmutableBytesWritable rowKeyPtr, byte[] regionStartKey, byte[] regionEndKey, boolean convertArrayColToKeyValueCol) {
ImmutableBytesWritable ptr = new ImmutableBytesWritable();
boolean prependRegionStartKey = isLocalIndex && regionStartKey != null;
boolean isIndexSalted = !isLocalIndex && nIndexSaltBuckets > 0;
@@ -524,6 +533,9 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
SortOrder dataSortOrder;
if (dataPkPosition[i] == EXPRESSION_NOT_PRESENT) {
Expression expression = expressionIterator.next();
+ if (convertArrayColToKeyValueCol) {
+ expression = expression.accept(new ReplaceArrayColumnWithKeyValueColumnExpressionVisitor());
+ }
dataColumnType = expression.getDataType();
dataSortOrder = expression.getSortOrder();
isNullable = expression.isNullable();
@@ -856,11 +868,11 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
return indexRowKeySchema;
}
- public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey) throws IOException {
+ public Put buildUpdateMutation(KeyValueBuilder kvBuilder, ValueGetter valueGetter, ImmutableBytesWritable dataRowKeyPtr, long ts, byte[] regionStartKey, byte[] regionEndKey, boolean convertArrayColToKeyValueCol) throws IOException {
Put put = null;
// New row being inserted: add the empty key value
if (valueGetter.getLatestValue(dataEmptyKeyValueRef) == null) {
- byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey);
+ byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey, convertArrayColToKeyValueCol);
put = new Put(indexRowKey);
// add the keyvalue for the empty row
put.add(kvBuilder.buildPut(new ImmutableBytesPtr(indexRowKey),
@@ -869,21 +881,72 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
emptyKeyValueQualifierPtr));
put.setDurability(!indexWALDisabled ? Durability.USE_DEFAULT : Durability.SKIP_WAL);
}
- for (ColumnReference ref : this.getCoveredColumns()) {
- //FIXME: samarth figure out a backward compatible way to handle this as coveredcolumnsmap won't be availble for older phoenix clients.
- ColumnReference indexColRef = this.coveredColumnsMap.get(ref);
- ImmutableBytesPtr cq = indexColRef.getQualifierWritable();
- ImmutableBytesPtr cf = indexColRef.getFamilyWritable();
- ImmutableBytesWritable value = valueGetter.getLatestValue(ref);
- byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey);
- ImmutableBytesPtr rowKey = new ImmutableBytesPtr(indexRowKey);
- if (value != null) {
+ byte[] indexRowKey = this.buildRowKey(valueGetter, dataRowKeyPtr, regionStartKey, regionEndKey, convertArrayColToKeyValueCol);
+ ImmutableBytesPtr rowKey = new ImmutableBytesPtr(indexRowKey);
+ if (storeColsInSingleCell) {
+ // map from column family to list of columns (for covered columns)
+ Map<String, List<ColumnReference>> familyToColListMap = Maps.newHashMap();
+ for (ColumnReference ref : this.getCoveredColumns()) {
+ String cf = Bytes.toString(ref.getFamily());
+ if (!familyToColListMap.containsKey(cf)) {
+ familyToColListMap.put(cf, Lists.<ColumnReference>newArrayList());
+ }
+ familyToColListMap.get(cf).add(ref);
+ }
+ // iterate over each column family and create a byte[] containing all the columns
+ for (Entry<String, List<ColumnReference>> entry : familyToColListMap.entrySet()) {
+ byte[] columnFamily = entry.getKey().getBytes();
+ List<ColumnReference> colRefs = entry.getValue();
+ int maxIndex = Integer.MIN_VALUE;
+ // find the max col qualifier
+ for (ColumnReference colRef : colRefs) {
+ byte[] qualifier = this.coveredColumnsMap.get(colRef).getQualifier();
+ maxIndex = Math.max(maxIndex, PInteger.INSTANCE.getCodec().decodeInt(qualifier, 0, SortOrder.getDefault()));
+ }
+ byte[][] colValues = new byte[maxIndex+1][];
+ // set the values of the columns
+ for (ColumnReference colRef : colRefs) {
+ ImmutableBytesWritable value = valueGetter.getLatestValue(colRef);
+ if (value != null) {
+ byte[] qualifier = this.coveredColumnsMap.get(colRef).getQualifier();
+ int index = PInteger.INSTANCE.getCodec().decodeInt(qualifier, 0, SortOrder.getDefault());
+ colValues[index] = value.get();
+ }
+ }
+
+ List<Expression> children = Lists.newArrayListWithExpectedSize(colRefs.size());
+ // create an expression list with all the columns
+ for (int j=0; j<colValues.length; ++j) {
+ children.add(new LiteralExpression(colValues[j]==null ? ByteUtil.EMPTY_BYTE_ARRAY : colValues[j] ));
+ }
+ // we use ArrayConstructorExpression to serialize multiple columns into a single byte[]
+ // construct the ArrayConstructorExpression with a variable length data type (PVarchar) since columns can be of fixed or variable length
+ ArrayConstructorExpression arrayExpression = new ArrayConstructorExpression(children, PVarchar.INSTANCE, rowKeyOrderOptimizable);
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ arrayExpression.evaluate(new BaseTuple() {}, ptr);
if (put == null) {
put = new Put(indexRowKey);
put.setDurability(!indexWALDisabled ? Durability.USE_DEFAULT : Durability.SKIP_WAL);
}
+ ImmutableBytesPtr colFamilyPtr = new ImmutableBytesPtr(columnFamily);
//this is a little bit of extra work for installations that are running <0.94.14, but that should be rare and is a short-term set of wrappers - it shouldn't kill GC
- put.add(kvBuilder.buildPut(rowKey, cf, cq, ts, value));
+ put.add(kvBuilder.buildPut(rowKey, colFamilyPtr, colFamilyPtr, ts, ptr));
+ }
+ }
+ else {
+ for (ColumnReference ref : this.getCoveredColumns()) {
+ //FIXME: samarth figure out a backward compatible way to handle this as coveredcolumnsmap won't be availble for older phoenix clients.
+ ColumnReference indexColRef = this.coveredColumnsMap.get(ref);
+ ImmutableBytesPtr cq = indexColRef.getQualifierWritable();
+ ImmutableBytesPtr cf = indexColRef.getFamilyWritable();
+ ImmutableBytesWritable value = valueGetter.getLatestValue(ref);
+ if (value != null) {
+ if (put == null) {
+ put = new Put(indexRowKey);
+ put.setDurability(!indexWALDisabled ? Durability.USE_DEFAULT : Durability.SKIP_WAL);
+ }
+ put.add(kvBuilder.buildPut(rowKey, cf, cq, ts, value));
+ }
}
}
return put;
@@ -962,7 +1025,7 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
@SuppressWarnings("deprecation")
public Delete buildDeleteMutation(KeyValueBuilder kvBuilder, ValueGetter oldState, ImmutableBytesWritable dataRowKeyPtr, Collection<KeyValue> pendingUpdates, long ts, byte[] regionStartKey, byte[] regionEndKey) throws IOException {
- byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr, regionStartKey, regionEndKey);
+ byte[] indexRowKey = this.buildRowKey(oldState, dataRowKeyPtr, regionStartKey, regionEndKey, false);
// Delete the entire row if any of the indexed columns changed
DeleteType deleteType = null;
if (oldState == null || (deleteType=getDeleteTypeOrNull(pendingUpdates)) != null || hasIndexedColumnChanged(oldState, pendingUpdates)) { // Deleting the entire row
@@ -1264,6 +1327,14 @@ public class IndexMaintainer implements Writable, Iterable<ColumnReference> {
}
return null;
}
+ @Override
+ public Void visit(ArrayColumnExpression expression) {
+ KeyValueColumnExpression colExpression = expression.getArrayExpression();
+ if (indexedColumns.add(new ColumnReference(colExpression.getColumnFamily(), colExpression.getColumnQualifier()))) {
+ indexedColumnTypes.add(colExpression.getDataType());
+ }
+ return null;
+ }
};
expression.accept(visitor);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
index 9d2955b..b1454b7 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixIndexCodec.java
@@ -74,7 +74,7 @@ public class PhoenixIndexCodec extends BaseIndexCodec {
indexUpdate.setTable(maintainer.isLocalIndex() ? state.getEnvironment().getRegion()
.getTableDesc().getName() : maintainer.getIndexTableName());
Put put = maintainer.buildUpdateMutation(KV_BUILDER, valueGetter, ptr, state.getCurrentTimestamp(), env
- .getRegion().getRegionInfo().getStartKey(), env.getRegion().getRegionInfo().getEndKey());
+ .getRegion().getRegionInfo().getStartKey(), env.getRegion().getRegionInfo().getEndKey(), false);
indexUpdate.setUpdate(put);
indexUpdates.add(indexUpdate);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 496b3b0..f6b929e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -52,6 +52,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.hadoop.hbase.HConstants;
+
import javax.management.Query;
import org.apache.hadoop.hbase.HRegionInfo;
@@ -87,6 +88,7 @@ import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PColumnFamily;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.schema.PTable.ViewType;
import org.apache.phoenix.schema.StaleRegionBoundaryCacheException;
import org.apache.phoenix.schema.TableRef;
@@ -260,7 +262,8 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
private static Pair<Integer, Integer> getMinMaxQualifiers(Scan scan, StatementContext context) {
PTable table = context.getCurrentTable().getTable();
- checkArgument(EncodedColumnsUtil.usesEncodedColumnNames(table), "Method should only be used for tables using encoded column names");
+ StorageScheme storageScheme = table.getStorageScheme();
+ checkArgument(EncodedColumnsUtil.usesEncodedColumnNames(storageScheme), "Method should only be used for tables using encoded column names");
Integer minQualifier = null;
Integer maxQualifier = null;
boolean emptyKVProjected = false;
@@ -410,7 +413,7 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
// the ExplicitColumnTracker not to be used, though.
if (!statement.isAggregate() && filteredColumnNotInProjection) {
ScanUtil.andFilterAtEnd(scan, new ColumnProjectionFilter(SchemaUtil.getEmptyColumnFamily(table),
- columnsTracker, conditionOnlyCfs, EncodedColumnsUtil.usesEncodedColumnNames(table)));
+ columnsTracker, conditionOnlyCfs, EncodedColumnsUtil.usesEncodedColumnNames(table.getStorageScheme())));
}
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index f373de2..bb3306e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -94,7 +94,6 @@ import org.apache.tephra.TransactionManager;
import org.apache.tephra.TransactionSystemClient;
import org.apache.tephra.inmemory.InMemoryTxSystemClient;
-
/**
*
* Implementation of ConnectionQueryServices used in testing where no connection to
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
index 544fb20..ed0c711 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/ColumnRef.java
@@ -18,10 +18,12 @@
package org.apache.phoenix.schema;
import org.apache.http.annotation.Immutable;
+import org.apache.phoenix.expression.ArrayColumnExpression;
import org.apache.phoenix.expression.ColumnExpression;
import org.apache.phoenix.expression.KeyValueColumnExpression;
import org.apache.phoenix.expression.ProjectedColumnExpression;
import org.apache.phoenix.expression.RowKeyColumnExpression;
+import org.apache.phoenix.schema.PTable.StorageScheme;
import org.apache.phoenix.util.EncodedColumnsUtil;
import org.apache.phoenix.util.SchemaUtil;
@@ -110,7 +112,10 @@ public class ColumnRef {
return new ProjectedColumnExpression(column, table, displayName);
}
- return new KeyValueColumnExpression(column, displayName, EncodedColumnsUtil.usesEncodedColumnNames(table));
+ if (table.getStorageScheme() == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL) {
+ return new ArrayColumnExpression(column, displayName, EncodedColumnsUtil.usesEncodedColumnNames(table.getStorageScheme()));
+ }
+ return new KeyValueColumnExpression(column, displayName, EncodedColumnsUtil.usesEncodedColumnNames(table.getStorageScheme()));
}
public ColumnRef cloneAtTimestamp(long timestamp) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
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 95e0f4f..18176b9 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
@@ -1805,7 +1805,7 @@ public class MetaDataClient {
updateCacheFrequency = updateCacheFrequencyProp;
}
String autoPartitionSeq = (String) TableProperty.AUTO_PARTITION_SEQ.getValue(tableProps);
-
+
Boolean storeNullsProp = (Boolean) TableProperty.STORE_NULLS.getValue(tableProps);
if (storeNullsProp == null) {
if (parent == null) {
@@ -1827,7 +1827,7 @@ public class MetaDataClient {
if (transactionalProp == null) {
transactional = connection.getQueryServices().getProps().getBoolean(
QueryServices.DEFAULT_TABLE_ISTRANSACTIONAL_ATTRIB,
- QueryServicesOptions.DEFAULT_TRANSACTIONAL);
+ QueryServicesOptions.DEFAULT_TABLE_ISTRANSACTIONAL);
} else {
transactional = transactionalProp;
}
@@ -2037,7 +2037,7 @@ public class MetaDataClient {
*/
viewPhysicalTable = connection.getTable(new PTableKey(null, physicalNames.get(0).getString()));
storageScheme = viewPhysicalTable.getStorageScheme();
- if (storageScheme == StorageScheme.ENCODED_COLUMN_NAMES) {
+ if (EncodedColumnsUtil.usesEncodedColumnNames(viewPhysicalTable)) {
cqCounter = viewPhysicalTable.getEncodedCQCounter();
}
}
@@ -2078,10 +2078,15 @@ public class MetaDataClient {
storageScheme = parent.getStorageScheme();
} else if (tableExists) {
storageScheme = StorageScheme.NON_ENCODED_COLUMN_NAMES;
+ } else if (isImmutableRows) {
+// storageScheme = StorageScheme.NON_ENCODED_COLUMN_NAMES;
+ storageScheme = StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL;
+ // since we are storing all columns of a column family in a single key value we can't use deletes to store nulls
+ storeNulls = true;
} else {
storageScheme = StorageScheme.ENCODED_COLUMN_NAMES;
}
- cqCounter = storageScheme == StorageScheme.ENCODED_COLUMN_NAMES ? new EncodedCQCounter() : NULL_COUNTER;
+ cqCounter = storageScheme != StorageScheme.ENCODED_COLUMN_NAMES ? new EncodedCQCounter() : NULL_COUNTER;
}
Map<String, Integer> changedCqCounters = new HashMap<>(colDefs.size());
@@ -2531,7 +2536,7 @@ public class MetaDataClient {
}
private static boolean incrementEncodedCQCounter(StorageScheme storageScheme, ColumnDef colDef) {
- return storageScheme == StorageScheme.ENCODED_COLUMN_NAMES && !colDef.isPK();
+ return storageScheme != StorageScheme.NON_ENCODED_COLUMN_NAMES && !colDef.isPK();
}
private byte[][] getSplitKeys(List<HRegionLocation> allTableRegions) {
@@ -2951,7 +2956,6 @@ public class MetaDataClient {
PName tenantId = connection.getTenantId();
String schemaName = table.getSchemaName().getString();
String tableName = table.getTableName().getString();
-
Boolean isImmutableRowsProp = null;
Boolean multiTenantProp = null;
Boolean disableWALProp = null;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
index ecef105..b9f3eaf 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
@@ -166,7 +166,8 @@ public interface PTable extends PMetaDataEntity {
public enum StorageScheme {
ENCODED_COLUMN_NAMES((byte)1),
- NON_ENCODED_COLUMN_NAMES((byte)2);
+ NON_ENCODED_COLUMN_NAMES((byte)2),
+ COLUMNS_STORED_IN_SINGLE_CELL((byte)3);
private final byte[] byteValue;
private final byte serializedValue;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
index dcc0e08..fee39ce 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PTableImpl.java
@@ -24,6 +24,7 @@ import static org.apache.phoenix.schema.SaltingUtil.SALTING_COLUMN;
import java.io.IOException;
import java.sql.SQLException;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -42,6 +43,9 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.coprocessor.generated.PTableProtos;
import org.apache.phoenix.exception.DataExceedsCapacityException;
+import org.apache.phoenix.expression.ArrayConstructorExpression;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.LiteralExpression;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
import org.apache.phoenix.index.IndexMaintainer;
@@ -49,12 +53,14 @@ import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.protobuf.ProtobufUtil;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.RowKeySchema.RowKeySchemaBuilder;
+import org.apache.phoenix.schema.tuple.BaseTuple;
import org.apache.phoenix.schema.types.PBinary;
import org.apache.phoenix.schema.types.PChar;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PDouble;
import org.apache.phoenix.schema.types.PFloat;
import org.apache.phoenix.schema.types.PInteger;
+import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.types.PVarchar;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.EncodedColumnsUtil;
@@ -421,7 +427,7 @@ public class PTableImpl implements PTable {
PColumn[] allColumns;
this.columnsByName = ArrayListMultimap.create(columns.size(), 1);
- this.kvColumnsByEncodedColumnNames = (storageScheme == StorageScheme.ENCODED_COLUMN_NAMES ? ArrayListMultimap.<Integer, PColumn>create(columns.size(), 1) : null);
+ this.kvColumnsByEncodedColumnNames = (EncodedColumnsUtil.usesEncodedColumnNames(storageScheme) ? ArrayListMultimap.<Integer, PColumn>create(columns.size(), 1) : null);
int numPKColumns = 0;
if (bucketNum != null) {
// Add salt column to allColumns and pkColumns, but don't add to
@@ -527,7 +533,7 @@ public class PTableImpl implements PTable {
.orderedBy(Bytes.BYTES_COMPARATOR);
for (int i = 0; i < families.length; i++) {
Map.Entry<PName,List<PColumn>> entry = iterator.next();
- PColumnFamily family = new PColumnFamilyImpl(entry.getKey(), entry.getValue(), storageScheme == StorageScheme.ENCODED_COLUMN_NAMES);
+ PColumnFamily family = new PColumnFamilyImpl(entry.getKey(), entry.getValue(), EncodedColumnsUtil.usesEncodedColumnNames(storageScheme));
families[i] = family;
familyByString.put(family.getName().getString(), family);
familyByBytes.put(family.getName().getBytes(), family);
@@ -688,7 +694,7 @@ public class PTableImpl implements PTable {
}
private PRow newRow(KeyValueBuilder builder, long ts, ImmutableBytesWritable key, int i, byte[]... values) {
- PRow row = new PRowImpl(builder, key, ts, getBucketNum());
+ PRow row = new PRowImpl(builder, key, ts, getBucketNum(), values.length);
if (i < values.length) {
for (PColumnFamily family : getColumnFamilies()) {
for (PColumn column : family.getColumns()) {
@@ -781,8 +787,10 @@ public class PTableImpl implements PTable {
private Delete unsetValues;
private Mutation deleteRow;
private final long ts;
+ // map from column name to value
+ private Map<PColumn, byte[]> columnToValueMap;
- public PRowImpl(KeyValueBuilder kvBuilder, ImmutableBytesWritable key, long ts, Integer bucketNum) {
+ public PRowImpl(KeyValueBuilder kvBuilder, ImmutableBytesWritable key, long ts, Integer bucketNum, int numColumns) {
this.kvBuilder = kvBuilder;
this.ts = ts;
if (bucketNum != null) {
@@ -792,7 +800,7 @@ public class PTableImpl implements PTable {
this.keyPtr = new ImmutableBytesPtr(key);
this.key = ByteUtil.copyKeyBytesIfNecessary(key);
}
-
+ this.columnToValueMap = Maps.newHashMapWithExpectedSize(numColumns);
newMutations();
}
@@ -814,8 +822,43 @@ public class PTableImpl implements PTable {
// Include only deleteRow mutation if present because it takes precedence over all others
mutations.add(deleteRow);
} else {
+ // store all columns for a given column family in a single cell instead of one column per cell in order to improve write performance
+ if (storageScheme == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL) {
+ Put put = new Put(this.key);
+ if (isWALDisabled()) {
+ put.setDurability(Durability.SKIP_WAL);
+ }
+ // the setValues Put contains one cell per column, we need to convert it to a Put that contains a cell with all columns for a given column family
+ for (PColumnFamily family : families) {
+ byte[] columnFamily = family.getName().getBytes();
+ Collection<PColumn> columns = family.getColumns();
+ int maxEncodedColumnQualifier = Integer.MIN_VALUE;
+ for (PColumn column : columns) {
+ maxEncodedColumnQualifier = Math.max(maxEncodedColumnQualifier, column.getEncodedColumnQualifier());
+ }
+ byte[][] colValues = new byte[maxEncodedColumnQualifier+1][];
+ for (PColumn column : columns) {
+ colValues[column.getEncodedColumnQualifier()] = columnToValueMap.get(column);
+ }
+
+ List<Expression> children = Lists.newArrayListWithExpectedSize(columns.size());
+ // create an expression list with all the columns
+ for (int i=0; i<colValues.length; ++i) {
+ children.add(new LiteralExpression(colValues[i]==null ? ByteUtil.EMPTY_BYTE_ARRAY : colValues[i] ));
+ }
+ // we use ArrayConstructorExpression to serialize multiple columns into a single byte[]
+ // construct the ArrayConstructorExpression with a variable length data type since columns can be of fixed or variable length
+ ArrayConstructorExpression arrayExpression = new ArrayConstructorExpression(children, PVarbinary.INSTANCE, rowKeyOrderOptimizable);
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ arrayExpression.evaluate(new BaseTuple() {}, ptr);
+ ImmutableBytesPtr colFamilyPtr = new ImmutableBytesPtr(columnFamily);
+ addQuietly(put, kvBuilder, kvBuilder.buildPut(keyPtr,
+ colFamilyPtr, colFamilyPtr, ts, ptr));
+ }
+ setValues = put;
+ }
// Because we cannot enforce a not null constraint on a KV column (since we don't know if the row exists when
- // we upsert it), se instead add a KV that is always emtpy. This allows us to imitate SQL semantics given the
+ // we upsert it), so instead add a KV that is always empty. This allows us to imitate SQL semantics given the
// way HBase works.
Pair<byte[], byte[]> emptyKvInfo = EncodedColumnsUtil.getEmptyKeyValueInfo(PTableImpl.this);
addQuietly(setValues, kvBuilder, kvBuilder.buildPut(keyPtr,
@@ -870,18 +913,18 @@ public class PTableImpl implements PTable {
} else {
ImmutableBytesWritable ptr = new ImmutableBytesWritable(byteValue == null ?
HConstants.EMPTY_BYTE_ARRAY : byteValue);
- Integer maxLength = column.getMaxLength();
- if (!isNull && type.isFixedWidth() && maxLength != null) {
- if (ptr.getLength() < maxLength) {
- type.pad(ptr, maxLength, column.getSortOrder());
- } else if (ptr.getLength() > maxLength) {
- throw new DataExceedsCapacityException(name.getString() + "." + column.getName().getString() + " may not exceed " + maxLength + " bytes (" + type.toObject(byteValue) + ")");
- }
+ SchemaUtil.padData(name.getString(), column, ptr);
+ // store all columns for a given column family in a single cell instead of one column per cell in order to improve write performance
+ // we don't need to do anything with unsetValues as it is only used when storeNulls is false, storeNulls is always true when storeColsInSingleCell is true
+ if (storageScheme == StorageScheme.COLUMNS_STORED_IN_SINGLE_CELL) {
+ columnToValueMap.put(column, ptr.get());
+ }
+ else {
+ removeIfPresent(unsetValues, family, qualifier);
+ addQuietly(setValues, kvBuilder, kvBuilder.buildPut(keyPtr,
+ column.getFamilyName().getBytesPtr(), qualifierPtr,
+ ts, ptr));
}
- removeIfPresent(unsetValues, family, qualifier);
- addQuietly(setValues, kvBuilder, kvBuilder.buildPut(keyPtr,
- column.getFamilyName().getBytesPtr(), qualifierPtr,
- ts, ptr));
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/TableProperty.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/TableProperty.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/TableProperty.java
index d5d0b84..6b55756 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/TableProperty.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/TableProperty.java
@@ -51,7 +51,7 @@ public enum TableProperty {
STORE_NULLS(PhoenixDatabaseMetaData.STORE_NULLS, COLUMN_FAMILY_NOT_ALLOWED_TABLE_PROPERTY, true, false),
TRANSACTIONAL(PhoenixDatabaseMetaData.TRANSACTIONAL, COLUMN_FAMILY_NOT_ALLOWED_TABLE_PROPERTY, true, false),
-
+
UPDATE_CACHE_FREQUENCY(PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY, true, true) {
@Override
public Object getValue(Object value) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/920be8fe/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/BaseTuple.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/BaseTuple.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/BaseTuple.java
index 92371e7..8028eb2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/BaseTuple.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/tuple/BaseTuple.java
@@ -20,9 +20,39 @@ package org.apache.phoenix.schema.tuple;
import java.util.List;
import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
public abstract class BaseTuple implements Tuple {
+ @Override
+ public int size() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean isImmutable() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void getKey(ImmutableBytesWritable ptr) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Cell getValue(int index) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Cell getValue(byte [] family, byte [] qualifier) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public boolean getValue(byte [] family, byte [] qualifier, ImmutableBytesWritable ptr) {
+ throw new UnsupportedOperationException();
+ }
@Override
public long getSequenceValue(int index) {