You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ra...@apache.org on 2014/07/16 06:33:01 UTC
[3/6] Phoenix-933 Local index support to Phoenix (Rajesh)
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
index cb6711a..9cb47b9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/generated/PTableProtos.java
@@ -2535,6 +2535,16 @@ public final class PTableProtos {
* <code>optional int32 viewIndexId = 21;</code>
*/
int getViewIndexId();
+
+ // optional bytes indexType = 22;
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ boolean hasIndexType();
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ com.google.protobuf.ByteString getIndexType();
}
/**
* Protobuf type {@code PTable}
@@ -2710,6 +2720,11 @@ public final class PTableProtos {
viewIndexId_ = input.readInt32();
break;
}
+ case 178: {
+ bitField0_ |= 0x00020000;
+ indexType_ = input.readBytes();
+ break;
+ }
}
}
} catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -3192,6 +3207,22 @@ public final class PTableProtos {
return viewIndexId_;
}
+ // optional bytes indexType = 22;
+ public static final int INDEXTYPE_FIELD_NUMBER = 22;
+ private com.google.protobuf.ByteString indexType_;
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public boolean hasIndexType() {
+ return ((bitField0_ & 0x00020000) == 0x00020000);
+ }
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public com.google.protobuf.ByteString getIndexType() {
+ return indexType_;
+ }
+
private void initFields() {
schemaNameBytes_ = com.google.protobuf.ByteString.EMPTY;
tableNameBytes_ = com.google.protobuf.ByteString.EMPTY;
@@ -3214,6 +3245,7 @@ public final class PTableProtos {
physicalNames_ = java.util.Collections.emptyList();
tenantId_ = com.google.protobuf.ByteString.EMPTY;
viewIndexId_ = 0;
+ indexType_ = com.google.protobuf.ByteString.EMPTY;
}
private byte memoizedIsInitialized = -1;
public final boolean isInitialized() {
@@ -3344,6 +3376,9 @@ public final class PTableProtos {
if (((bitField0_ & 0x00010000) == 0x00010000)) {
output.writeInt32(21, viewIndexId_);
}
+ if (((bitField0_ & 0x00020000) == 0x00020000)) {
+ output.writeBytes(22, indexType_);
+ }
getUnknownFields().writeTo(output);
}
@@ -3442,6 +3477,10 @@ public final class PTableProtos {
size += com.google.protobuf.CodedOutputStream
.computeInt32Size(21, viewIndexId_);
}
+ if (((bitField0_ & 0x00020000) == 0x00020000)) {
+ size += com.google.protobuf.CodedOutputStream
+ .computeBytesSize(22, indexType_);
+ }
size += getUnknownFields().getSerializedSize();
memoizedSerializedSize = size;
return size;
@@ -3558,6 +3597,11 @@ public final class PTableProtos {
result = result && (getViewIndexId()
== other.getViewIndexId());
}
+ result = result && (hasIndexType() == other.hasIndexType());
+ if (hasIndexType()) {
+ result = result && getIndexType()
+ .equals(other.getIndexType());
+ }
result = result &&
getUnknownFields().equals(other.getUnknownFields());
return result;
@@ -3655,6 +3699,10 @@ public final class PTableProtos {
hash = (37 * hash) + VIEWINDEXID_FIELD_NUMBER;
hash = (53 * hash) + getViewIndexId();
}
+ if (hasIndexType()) {
+ hash = (37 * hash) + INDEXTYPE_FIELD_NUMBER;
+ hash = (53 * hash) + getIndexType().hashCode();
+ }
hash = (29 * hash) + getUnknownFields().hashCode();
memoizedHashCode = hash;
return hash;
@@ -3821,6 +3869,8 @@ public final class PTableProtos {
bitField0_ = (bitField0_ & ~0x00080000);
viewIndexId_ = 0;
bitField0_ = (bitField0_ & ~0x00100000);
+ indexType_ = com.google.protobuf.ByteString.EMPTY;
+ bitField0_ = (bitField0_ & ~0x00200000);
return this;
}
@@ -3949,6 +3999,10 @@ public final class PTableProtos {
to_bitField0_ |= 0x00010000;
}
result.viewIndexId_ = viewIndexId_;
+ if (((from_bitField0_ & 0x00200000) == 0x00200000)) {
+ to_bitField0_ |= 0x00020000;
+ }
+ result.indexType_ = indexType_;
result.bitField0_ = to_bitField0_;
onBuilt();
return result;
@@ -4106,6 +4160,9 @@ public final class PTableProtos {
if (other.hasViewIndexId()) {
setViewIndexId(other.getViewIndexId());
}
+ if (other.hasIndexType()) {
+ setIndexType(other.getIndexType());
+ }
this.mergeUnknownFields(other.getUnknownFields());
return this;
}
@@ -5608,6 +5665,42 @@ public final class PTableProtos {
return this;
}
+ // optional bytes indexType = 22;
+ private com.google.protobuf.ByteString indexType_ = com.google.protobuf.ByteString.EMPTY;
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public boolean hasIndexType() {
+ return ((bitField0_ & 0x00200000) == 0x00200000);
+ }
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public com.google.protobuf.ByteString getIndexType() {
+ return indexType_;
+ }
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public Builder setIndexType(com.google.protobuf.ByteString value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ bitField0_ |= 0x00200000;
+ indexType_ = value;
+ onChanged();
+ return this;
+ }
+ /**
+ * <code>optional bytes indexType = 22;</code>
+ */
+ public Builder clearIndexType() {
+ bitField0_ = (bitField0_ & ~0x00200000);
+ indexType_ = getDefaultInstance().getIndexType();
+ onChanged();
+ return this;
+ }
+
// @@protoc_insertion_point(builder_scope:PTable)
}
@@ -5650,7 +5743,7 @@ public final class PTableProtos {
" \002(\005\022\021\n\tsortOrder\030\010 \002(\005\022\021\n\tarraySize\030\t \001" +
"(\005\022\024\n\014viewConstant\030\n \001(\014\022\026\n\016viewReferenc" +
"ed\030\013 \001(\010\"*\n\013PTableStats\022\013\n\003key\030\001 \002(\t\022\016\n\006" +
- "values\030\002 \003(\014\"\367\003\n\006PTable\022\027\n\017schemaNameByt" +
+ "values\030\002 \003(\014\"\212\004\n\006PTable\022\027\n\017schemaNameByt" +
"es\030\001 \002(\014\022\026\n\016tableNameBytes\030\002 \002(\014\022\036\n\ttabl" +
"eType\030\003 \002(\0162\013.PTableType\022\022\n\nindexState\030\004",
" \001(\t\022\026\n\016sequenceNumber\030\005 \002(\003\022\021\n\ttimeStam" +
@@ -5662,11 +5755,11 @@ public final class PTableProtos {
"milyName\030\016 \001(\014\022\022\n\ndisableWAL\030\017 \002(\010\022\023\n\013mu" +
"ltiTenant\030\020 \002(\010\022\020\n\010viewType\030\021 \001(\014\022\025\n\rvie" +
"wStatement\030\022 \001(\014\022\025\n\rphysicalNames\030\023 \003(\014\022" +
- "\020\n\010tenantId\030\024 \001(\014\022\023\n\013viewIndexId\030\025 \001(\005*A",
- "\n\nPTableType\022\n\n\006SYSTEM\020\000\022\010\n\004USER\020\001\022\010\n\004VI" +
- "EW\020\002\022\t\n\005INDEX\020\003\022\010\n\004JOIN\020\004B@\n(org.apache." +
- "phoenix.coprocessor.generatedB\014PTablePro" +
- "tosH\001\210\001\001\240\001\001"
+ "\020\n\010tenantId\030\024 \001(\014\022\023\n\013viewIndexId\030\025 \001(\005\022\021",
+ "\n\tindexType\030\026 \001(\014*A\n\nPTableType\022\n\n\006SYSTE" +
+ "M\020\000\022\010\n\004USER\020\001\022\010\n\004VIEW\020\002\022\t\n\005INDEX\020\003\022\010\n\004JO" +
+ "IN\020\004B@\n(org.apache.phoenix.coprocessor.g" +
+ "eneratedB\014PTableProtosH\001\210\001\001\240\001\001"
};
com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -5690,7 +5783,7 @@ public final class PTableProtos {
internal_static_PTable_fieldAccessorTable = new
com.google.protobuf.GeneratedMessage.FieldAccessorTable(
internal_static_PTable_descriptor,
- new java.lang.String[] { "SchemaNameBytes", "TableNameBytes", "TableType", "IndexState", "SequenceNumber", "TimeStamp", "PkNameBytes", "BucketNum", "Columns", "Indexes", "IsImmutableRows", "GuidePosts", "DataTableNameBytes", "DefaultFamilyName", "DisableWAL", "MultiTenant", "ViewType", "ViewStatement", "PhysicalNames", "TenantId", "ViewIndexId", });
+ new java.lang.String[] { "SchemaNameBytes", "TableNameBytes", "TableType", "IndexState", "SequenceNumber", "TimeStamp", "PkNameBytes", "BucketNum", "Columns", "Indexes", "IsImmutableRows", "GuidePosts", "DataTableNameBytes", "DefaultFamilyName", "DisableWAL", "MultiTenant", "ViewType", "ViewStatement", "PhysicalNames", "TenantId", "ViewIndexId", "IndexType", });
return null;
}
};
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
index 39b951d..4667051 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -211,7 +211,8 @@ public enum SQLExceptionCode {
INSUFFICIENT_MULTI_TENANT_COLUMNS(1040, "42Y96", "A MULTI_TENANT table must have two or more PK columns with the first column being NOT NULL and of type VARCHAR or CHAR."),
VIEW_WHERE_IS_CONSTANT(1045, "43A02", "WHERE clause in VIEW should not evaluate to a constant."),
CANNOT_UPDATE_VIEW_COLUMN(1046, "43A03", "Column updated in VIEW may not differ from value specified in WHERE clause."),
- TOO_MANY_VIEW_INDEXES(1047, "43A04", "Too many indexes have already been created on the physical table."),
+ TOO_MANY_INDEXES(1047, "43A04", "Too many indexes have already been created on the physical table."),
+ NO_LOCAL_INDEX_ON_TABLE_WITH_IMMUTABLE_ROWS(1048,"43A04","Local indexes aren't allowed on tables with immutable rows."),
/** Sequence related */
SEQUENCE_ALREADY_EXIST(1200, "42Z00", "Sequence already exists.", new Factory() {
@@ -250,6 +251,8 @@ public enum SQLExceptionCode {
SPLIT_POINT_NOT_CONSTANT(1105, "XCL04", "Split points must be constants."),
BATCH_EXCEPTION(1106, "XCL05", "Exception while executing batch."),
EXECUTE_UPDATE_WITH_NON_EMPTY_BATCH(1107, "XCL06", "An executeUpdate is prohibited when the batch is not empty. Use clearBatch to empty the batch first."),
+ CANNOT_SPLIT_LOCAL_INDEX(1108,"XCL07", "Local index may not be pre-split"),
+ CANNOT_SALT_LOCAL_INDEX(1109,"XCL08", "Local index may not be salted"),
/**
* Implementation defined class. Phoenix internal error. (errorcode 20, sqlstate INT).
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
index d0c97bd..be96956 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
@@ -17,26 +17,47 @@
*/
package org.apache.phoenix.execute;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
import java.sql.ParameterMetaData;
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.compile.ExplainPlan;
import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.compile.FromCompiler;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.RowProjector;
import org.apache.phoenix.compile.ScanRanges;
import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
+import org.apache.phoenix.expression.ProjectedColumnExpression;
+import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.iterate.DelegateResultIterator;
import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.parse.ParseNodeFactory;
+import org.apache.phoenix.parse.TableName;
+import org.apache.phoenix.schema.KeyValueSchema;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.IndexUtil;
import org.apache.phoenix.util.SQLCloseable;
import org.apache.phoenix.util.SQLCloseables;
import org.apache.phoenix.util.ScanUtil;
@@ -131,6 +152,8 @@ public abstract class BasicQueryPlan implements QueryPlan {
return ResultIterator.EMPTY_ITERATOR;
}
+ // Set miscellaneous scan attributes. This is the last chance to set them before we
+ // clone the scan for each parallelized chunk.
Scan scan = context.getScan();
// Set producer on scan so HBase server does round robin processing
//setProducer(scan);
@@ -150,6 +173,35 @@ public abstract class BasicQueryPlan implements QueryPlan {
}
ScanUtil.setTimeRange(scan, scn);
ScanUtil.setTenantId(scan, connection.getTenantId() == null ? null : connection.getTenantId().getBytes());
+ // Set local index related scan attributes.
+ if (context.getCurrentTable().getTable().getIndexType() == IndexType.LOCAL) {
+ ScanUtil.setLocalIndex(scan);
+ Set<PColumn> dataColumns = context.getDataColumns();
+ // If any data columns to join back from data table are present then we set following attributes
+ // 1. data columns to be projected and their key value schema.
+ // 2. index maintainer and view constants if exists to build data row key from index row key.
+ // 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.
+ serializeDataTableColumnsToJoin(scan, dataColumns);
+ KeyValueSchema schema = ProjectedColumnExpression.buildSchema(dataColumns);
+ // Set key value schema of the data columns.
+ serializeSchemaIntoScan(scan, schema);
+ String schemaName = context.getCurrentTable().getTable().getSchemaName().getString();
+ String parentTable = context.getCurrentTable().getTable().getParentTableName().getString();
+ final ParseNodeFactory FACTORY = new ParseNodeFactory();
+ TableRef dataTableRef =
+ FromCompiler.getResolver(
+ FACTORY.namedTable(null, TableName.create(schemaName, parentTable)),
+ context.getConnection()).resolveTable(schemaName, parentTable);
+ PTable dataTable = dataTableRef.getTable();
+ // Set index maintainer of the local index.
+ serializeIndexMaintainerIntoScan(scan, dataTable);
+ // Set view constants if exists.
+ serializeViewConstantsIntoScan(scan, dataTable);
+ }
+ }
ResultIterator iterator = newIterator();
return dependencies.isEmpty() ?
iterator : new DelegateResultIterator(iterator) {
@@ -164,6 +216,108 @@ public abstract class BasicQueryPlan implements QueryPlan {
};
}
+ private void serializeIndexMaintainerIntoScan(Scan scan, PTable dataTable) {
+ PName name = context.getCurrentTable().getTable().getName();
+ List<PTable> indexes = Lists.newArrayListWithExpectedSize(1);
+ for (PTable index : dataTable.getIndexes()) {
+ if (index.getName().equals(name) && index.getIndexType() == IndexType.LOCAL) {
+ indexes.add(index);
+ break;
+ }
+ }
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ IndexMaintainer.serialize(dataTable, ptr, indexes);
+ scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD, ByteUtil.copyKeyBytesIfNecessary(ptr));
+ }
+
+ private void serializeViewConstantsIntoScan(Scan scan, PTable dataTable) {
+ int dataPosOffset = (dataTable.getBucketNum() != null ? 1 : 0) + (dataTable.isMultiTenant() ? 1 : 0);
+ int nViewConstants = 0;
+ if (dataTable.getType() == PTableType.VIEW) {
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ List<PColumn> dataPkColumns = dataTable.getPKColumns();
+ for (int i = dataPosOffset; i < dataPkColumns.size(); i++) {
+ PColumn dataPKColumn = dataPkColumns.get(i);
+ if (dataPKColumn.getViewConstant() != null) {
+ nViewConstants++;
+ }
+ }
+ if (nViewConstants > 0) {
+ byte[][] viewConstants = new byte[nViewConstants][];
+ int j = 0;
+ for (int i = dataPosOffset; i < dataPkColumns.size(); i++) {
+ PColumn dataPkColumn = dataPkColumns.get(i);
+ if (dataPkColumn.getViewConstant() != null) {
+ if (IndexUtil.getViewConstantValue(dataPkColumn, ptr)) {
+ viewConstants[j++] = ByteUtil.copyKeyBytesIfNecessary(ptr);
+ } else {
+ throw new IllegalStateException();
+ }
+ }
+ }
+ serializeViewConstantsIntoScan(viewConstants, scan);
+ }
+ }
+ }
+
+ private void serializeViewConstantsIntoScan(byte[][] viewConstants, Scan scan) {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ try {
+ DataOutputStream output = new DataOutputStream(stream);
+ WritableUtils.writeVInt(output, viewConstants.length);
+ for (byte[] viewConstant : viewConstants) {
+ Bytes.writeByteArray(output, viewConstant);
+ }
+ scan.setAttribute(BaseScannerRegionObserver.VIEW_CONSTANTS, stream.toByteArray());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void serializeDataTableColumnsToJoin(Scan scan, Set<PColumn> dataColumns) {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ try {
+ DataOutputStream output = new DataOutputStream(stream);
+ WritableUtils.writeVInt(output, dataColumns.size());
+ for (PColumn column : dataColumns) {
+ Bytes.writeByteArray(output, column.getFamilyName().getBytes());
+ Bytes.writeByteArray(output, column.getName().getBytes());
+ }
+ scan.setAttribute(BaseScannerRegionObserver.DATA_TABLE_COLUMNS_TO_JOIN, stream.toByteArray());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ private void serializeSchemaIntoScan(Scan scan, KeyValueSchema schema) {
+ ByteArrayOutputStream stream = new ByteArrayOutputStream(schema.getEstimatedByteSize());
+ try {
+ DataOutputStream output = new DataOutputStream(stream);
+ schema.write(output);
+ scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_JOIN_SCHEMA, stream.toByteArray());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ stream.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
abstract protected ResultIterator newIterator() throws SQLException;
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
index a994067..f9af543 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
@@ -42,6 +42,7 @@ import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.SaltingUtil;
import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.util.ScanUtil;
import java.sql.SQLException;
@@ -115,7 +116,7 @@ public class ScanPlan extends BasicQueryPlan {
if (isOrdered) {
scanner = new MergeSortTopNResultIterator(iterators, limit, orderBy.getOrderByExpressions());
} else {
- if (isSalted &&
+ if ((isSalted || table.getIndexType() == IndexType.LOCAL) &&
(context.getConnection().getQueryServices().getProps().getBoolean(
QueryServices.ROW_KEY_ORDER_SALTED_TABLE_ATTRIB,
QueryServicesOptions.DEFAULT_ROW_KEY_ORDER_SALTED_TABLE) ||
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/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 7e4b9b6..27330d6 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
@@ -61,6 +61,7 @@ import org.apache.phoenix.expression.function.RoundDateExpression;
import org.apache.phoenix.expression.function.RoundDecimalExpression;
import org.apache.phoenix.expression.function.RoundFunction;
import org.apache.phoenix.expression.function.RoundTimestampExpression;
+import org.apache.phoenix.expression.function.SQLIndexTypeFunction;
import org.apache.phoenix.expression.function.SQLTableTypeFunction;
import org.apache.phoenix.expression.function.SQLViewTypeFunction;
import org.apache.phoenix.expression.function.SqlTypeNameFunction;
@@ -178,8 +179,8 @@ public enum ExpressionType {
ArrayAnyComparisonExpression(ArrayAnyComparisonExpression.class),
ArrayAllComparisonExpression(ArrayAllComparisonExpression.class),
InlineArrayElemRefExpression(InlineArrayElemRefExpression.class),
+ SQLIndexTypeFunction(SQLIndexTypeFunction.class),
ModulusExpression(ModulusExpression.class);
-
ExpressionType(Class<? extends Expression> clazz) {
this.clazz = clazz;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/expression/KeyValueColumnExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/KeyValueColumnExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/KeyValueColumnExpression.java
index 0e35bcb..4b5fdbb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/KeyValueColumnExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/KeyValueColumnExpression.java
@@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
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.util.SchemaUtil;
@@ -49,6 +50,12 @@ public class KeyValueColumnExpression extends ColumnExpression {
this(column, null);
}
+ public KeyValueColumnExpression(PDatum column, byte[] cf, byte[] cq) {
+ super(column);
+ this.cf = cf;
+ this.cq = cq;
+ }
+
public KeyValueColumnExpression(PColumn column, String displayName) {
super(column);
this.cf = column.getFamilyName().getBytes();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/expression/ProjectedColumnExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/ProjectedColumnExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/ProjectedColumnExpression.java
index dcac849..1c4232a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/ProjectedColumnExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/ProjectedColumnExpression.java
@@ -20,6 +20,7 @@ package org.apache.phoenix.expression;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
+import java.util.Collection;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.phoenix.expression.visitor.ExpressionVisitor;
@@ -49,12 +50,24 @@ public class ProjectedColumnExpression extends ColumnExpression {
this.displayName = displayName;
}
- private static KeyValueSchema buildSchema(PTable table) {
- KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0);
- for (PColumn column : table.getColumns()) {
- if (!SchemaUtil.isPKColumn(column)) {
- builder.addField(column);
- }
+ public ProjectedColumnExpression(PColumn column, Collection<PColumn> columns, int position, String displayName) {
+ super(column);
+ this.schema = buildSchema(columns);
+ this.bitSet = ValueBitSet.newInstance(schema);
+ this.position = position;
+ this.displayName = displayName;
+ }
+
+ private static KeyValueSchema buildSchema(PTable table) {
+ return buildSchema(table.getColumns());
+ }
+
+ public static KeyValueSchema buildSchema(Collection<PColumn> columns) {
+ KeyValueSchemaBuilder builder = new KeyValueSchemaBuilder(0);
+ for (PColumn column : columns) {
+ if (!SchemaUtil.isPKColumn(column)) {
+ builder.addField(column);
+ }
}
return builder.build();
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
index 6761733..7913ab8 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/RowKeyColumnExpression.java
@@ -41,6 +41,7 @@ public class RowKeyColumnExpression extends ColumnExpression {
private PDataType fromType;
private RowKeyValueAccessor accessor;
protected final String name;
+ private int offset;
public RowKeyColumnExpression() {
name = null; // Only on client
@@ -65,6 +66,15 @@ public class RowKeyColumnExpression extends ColumnExpression {
this(datum, accessor, fromType, datum.toString());
}
+ /**
+ * Used to set an offset to be skipped from the start of a the row key. Used by
+ * local indexing to skip the region start key bytes.
+ * @param offset the number of bytes to offset accesses to row key columns
+ */
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
public int getPosition() {
return accessor.getIndex();
}
@@ -94,7 +104,7 @@ public class RowKeyColumnExpression extends ColumnExpression {
@Override
public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
tuple.getKey(ptr);
- int offset = accessor.getOffset(ptr.get(), ptr.getOffset());
+ int offset = accessor.getOffset(ptr.get(), ptr.getOffset() + this.offset);
// Null is represented in the last expression of a multi-part key
// by the bytes not being present.
int maxOffset = ptr.getOffset() + ptr.getLength();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/expression/function/SQLIndexTypeFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/SQLIndexTypeFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/SQLIndexTypeFunction.java
new file mode 100644
index 0000000..617e977
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/SQLIndexTypeFunction.java
@@ -0,0 +1,77 @@
+/*
+ * 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.function;
+
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.parse.FunctionParseNode.Argument;
+import org.apache.phoenix.parse.FunctionParseNode.BuiltInFunction;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.PTable.IndexType;
+import org.apache.phoenix.schema.tuple.Tuple;
+
+
+/**
+ *
+ * Function used to get the SQL view type name from the serialized view type.
+ * Usage:
+ * SQLViewType('v') will return 'VIEW' based on
+ * {@link java.sql.DatabaseMetaData#getTableTypes()}
+ *
+ *
+ * @since 2.2
+ */
+@BuiltInFunction(name=SQLIndexTypeFunction.NAME, args= {
+ @Argument(allowedTypes=PDataType.UNSIGNED_TINYINT)} )
+public class SQLIndexTypeFunction extends ScalarFunction {
+ public static final String NAME = "SQLIndexType";
+
+ public SQLIndexTypeFunction() {
+ }
+
+ public SQLIndexTypeFunction(List<Expression> children) throws SQLException {
+ super(children);
+ }
+
+ @Override
+ public boolean evaluate(Tuple tuple, ImmutableBytesWritable ptr) {
+ Expression child = children.get(0);
+ if (!child.evaluate(tuple, ptr)) {
+ return false;
+ }
+ if (ptr.getLength() == 0) {
+ return true;
+ }
+ IndexType viewType = IndexType.fromSerializedValue(ptr.get()[ptr.getOffset()]);
+ ptr.set(viewType.getBytes());
+ return true;
+ }
+
+ @Override
+ public PDataType getDataType() {
+ return PDataType.VARCHAR;
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
new file mode 100644
index 0000000..a3e8fb0
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/visitor/RowKeyExpressionVisitor.java
@@ -0,0 +1,36 @@
+/*
+ * 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 org.apache.phoenix.expression.RowKeyColumnExpression;
+
+
+
+
+/**
+ *
+ * Implementation of ExpressionVisitor where only a RowKeyColumnExpression (i.e.
+ * a reference to a column that makes up the row key) is being visited,
+ *
+ *
+ * @since 0.1
+ */
+public abstract class RowKeyExpressionVisitor extends TraverseAllExpressionVisitor<Void> {
+ @Override
+ abstract public Void visit(RowKeyColumnExpression node);
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java b/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
index b627226..8172956 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/filter/SkipScanFilter.java
@@ -72,6 +72,7 @@ public class SkipScanFilter extends FilterBase implements Writable {
private byte[] endKey;
private int endKeyLength;
private boolean isDone;
+ private int offset;
private final ImmutableBytesWritable ptr = new ImmutableBytesWritable();
@@ -85,6 +86,10 @@ public class SkipScanFilter extends FilterBase implements Writable {
public SkipScanFilter(List<List<KeyRange>> slots, RowKeySchema schema) {
init(slots, schema);
}
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
private void init(List<List<KeyRange>> slots, RowKeySchema schema) {
for (List<KeyRange> ranges : slots) {
@@ -113,7 +118,7 @@ public class SkipScanFilter extends FilterBase implements Writable {
@Override
public ReturnCode filterKeyValue(Cell kv) {
- return navigate(kv.getRowArray(), kv.getRowOffset(),kv.getRowLength(),Terminate.AFTER);
+ return navigate(kv.getRowArray(), kv.getRowOffset() + offset,kv.getRowLength()- offset,Terminate.AFTER);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionSplitPolicy.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionSplitPolicy.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionSplitPolicy.java
new file mode 100644
index 0000000..0ee27e2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionSplitPolicy.java
@@ -0,0 +1,32 @@
+/*
+ * 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.hbase.index;
+
+import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
+
+/**
+ * Split policy for index regions to avoid split from external requests.
+ */
+public class IndexRegionSplitPolicy extends RegionSplitPolicy {
+
+ @Override
+ protected boolean shouldSplit() {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/balancer/IndexLoadBalancer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/balancer/IndexLoadBalancer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/balancer/IndexLoadBalancer.java
new file mode 100644
index 0000000..fc1963b
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/balancer/IndexLoadBalancer.java
@@ -0,0 +1,688 @@
+/**
+ * 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.hbase.index.balancer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.ClusterStatus;
+import org.apache.hadoop.hbase.HBaseIOException;
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.master.LoadBalancer;
+import org.apache.hadoop.hbase.master.MasterServices;
+import org.apache.hadoop.hbase.master.RegionPlan;
+import org.apache.hadoop.hbase.master.RegionState;
+import org.apache.hadoop.hbase.master.RegionStates;
+import org.apache.hadoop.hbase.master.balancer.StochasticLoadBalancer;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.util.ReflectionUtils;
+
+/**
+ * <p>This class is an extension of the load balancer class.
+ * It allows to co-locate the regions of the user table and the regions of corresponding
+ * index table if any.</p>
+ *
+ * </>roundRobinAssignment, retainAssignment -> index regions will follow the actual table regions.
+ * randomAssignment, balancerCluster -> either index table or actual table region(s) will follow
+ * each other based on which ever comes first.</p>
+ *
+ * <p>In case of master failover there is a chance that the znodes of the index
+ * table and actual table are left behind. Then in that scenario we may get randomAssignment for
+ * either the actual table region first or the index table region first.</p>
+ *
+ * <p>In case of balancing by table any table can balance first.</p>
+ *
+ */
+
+public class IndexLoadBalancer implements LoadBalancer {
+
+ private static final Log LOG = LogFactory.getLog(IndexLoadBalancer.class);
+
+ public static final byte[] PARENT_TABLE_KEY = Bytes.toBytes("PARENT_TABLE");
+
+ public static final String INDEX_BALANCER_DELEGATOR = "hbase.index.balancer.delegator.class";
+
+ private LoadBalancer delegator;
+
+ private MasterServices master;
+
+ private Configuration conf;
+
+ private ClusterStatus clusterStatus;
+
+ private static final Random RANDOM = new Random(EnvironmentEdgeManager.currentTimeMillis());
+
+ Map<TableName, TableName> userTableVsIndexTable = new HashMap<TableName, TableName>();
+
+ Map<TableName, TableName> indexTableVsUserTable = new HashMap<TableName, TableName>();
+
+ /**
+ * Maintains colocation information of user regions and corresponding index regions.
+ */
+ private Map<TableName, Map<ImmutableBytesWritable, ServerName>> colocationInfo =
+ new ConcurrentHashMap<TableName, Map<ImmutableBytesWritable, ServerName>>();
+
+ private Set<TableName> balancedTables = new HashSet<TableName>();
+
+ private boolean stopped = false;
+
+ @Override
+ public void initialize() throws HBaseIOException {
+ Class<? extends LoadBalancer> delegatorKlass =
+ conf.getClass(INDEX_BALANCER_DELEGATOR, StochasticLoadBalancer.class,
+ LoadBalancer.class);
+ this.delegator = ReflectionUtils.newInstance(delegatorKlass, conf);
+ this.delegator.setClusterStatus(clusterStatus);
+ this.delegator.setMasterServices(this.master);
+ this.delegator.initialize();
+ try {
+ populateTablesToColocate(this.master.getTableDescriptors().getAll());
+ } catch (IOException e) {
+ throw new HBaseIOException(e);
+ }
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public void setConf(Configuration configuration) {
+ this.conf = configuration;
+ }
+
+ @Override
+ public void setClusterStatus(ClusterStatus st) {
+ this.clusterStatus = st;
+ }
+
+ public Map<TableName, Map<ImmutableBytesWritable, ServerName>> getColocationInfo() {
+ return colocationInfo;
+ }
+
+ @Override
+ public void setMasterServices(MasterServices masterServices) {
+ this.master = masterServices;
+ }
+
+ @Override
+ public List<RegionPlan> balanceCluster(Map<ServerName, List<HRegionInfo>> clusterState)
+ throws HBaseIOException {
+ synchronized (this.colocationInfo) {
+ boolean balanceByTable = conf.getBoolean("hbase.master.loadbalance.bytable", false);
+ List<RegionPlan> regionPlans = null;
+
+ TableName tableName = null;
+ if (balanceByTable) {
+ Map<ImmutableBytesWritable, ServerName> tableKeys = null;
+ for (Entry<ServerName, List<HRegionInfo>> serverVsRegionList : clusterState
+ .entrySet()) {
+ ServerName sn = serverVsRegionList.getKey();
+ List<HRegionInfo> regionInfos = serverVsRegionList.getValue();
+ if (regionInfos.isEmpty()) {
+ continue;
+ }
+ if (!isTableColocated(regionInfos.get(0).getTable())) {
+ return this.delegator.balanceCluster(clusterState);
+ }
+ // Just get the table name from any one of the values in the regioninfo list
+ if (tableName == null) {
+ tableName = regionInfos.get(0).getTable();
+ tableKeys = this.colocationInfo.get(tableName);
+ }
+ // Check and modify the colocation info map based on values of cluster state
+ // because we
+ // will
+ // call balancer only when the cluster is in stable and reliable state.
+ if (tableKeys != null) {
+ for (HRegionInfo hri : regionInfos) {
+ updateServer(tableKeys, sn, hri);
+ }
+ }
+ }
+ // If user table is already balanced find the index table plans from the user table
+ // plans
+ // or vice verca.
+ TableName mappedTableName = getMappedTableToColocate(tableName);
+ if (balancedTables.contains(mappedTableName)) {
+ balancedTables.remove(mappedTableName);
+ regionPlans = new ArrayList<RegionPlan>();
+ return prepareRegionPlansForClusterState(clusterState, regionPlans);
+ } else {
+ balancedTables.add(tableName);
+ regionPlans = this.delegator.balanceCluster(clusterState);
+ if (regionPlans == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(tableName + " regions already balanced.");
+ }
+ return null;
+ } else {
+ updateRegionPlans(regionPlans);
+ return regionPlans;
+ }
+ }
+
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Seperating user tables and index tables regions of "
+ + "each region server in the cluster.");
+ }
+ Map<ServerName, List<HRegionInfo>> userClusterState =
+ new HashMap<ServerName, List<HRegionInfo>>();
+ Map<ServerName, List<HRegionInfo>> indexClusterState =
+ new HashMap<ServerName, List<HRegionInfo>>();
+ for (Entry<ServerName, List<HRegionInfo>> serverVsRegionList : clusterState
+ .entrySet()) {
+ ServerName sn = serverVsRegionList.getKey();
+ List<HRegionInfo> regionsInfos = serverVsRegionList.getValue();
+ List<HRegionInfo> idxRegionsToBeMoved = new ArrayList<HRegionInfo>();
+ List<HRegionInfo> userRegionsToBeMoved = new ArrayList<HRegionInfo>();
+ for (HRegionInfo hri : regionsInfos) {
+ if (hri.isMetaRegion()) {
+ continue;
+ }
+ tableName = hri.getTable();
+ // Check and modify the colocation info map based on values of cluster state
+ // because we
+ // will
+ // call balancer only when the cluster is in stable and reliable state.
+ if (isTableColocated(tableName)) {
+ // table name may change every time thats why always need to get table
+ // entries.
+ Map<ImmutableBytesWritable, ServerName> tableKeys =
+ this.colocationInfo.get(tableName);
+ if (tableKeys != null) {
+ updateServer(tableKeys, sn, hri);
+ }
+ }
+ if (indexTableVsUserTable.containsKey(tableName)) {
+ idxRegionsToBeMoved.add(hri);
+ continue;
+ }
+ userRegionsToBeMoved.add(hri);
+ }
+ // there may be dummy entries here if assignments by table is set
+ userClusterState.put(sn, userRegionsToBeMoved);
+ indexClusterState.put(sn, idxRegionsToBeMoved);
+ }
+
+ regionPlans = this.delegator.balanceCluster(userClusterState);
+ if (regionPlans == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("User region plan is null.");
+ }
+ regionPlans = new ArrayList<RegionPlan>();
+ } else {
+ updateRegionPlans(regionPlans);
+ }
+ return prepareRegionPlansForClusterState(indexClusterState, regionPlans);
+ }
+ }
+ }
+
+ private void updateServer(Map<ImmutableBytesWritable, ServerName> tableKeys, ServerName sn,
+ HRegionInfo hri) {
+ ImmutableBytesWritable startKey = new ImmutableBytesWritable(hri.getStartKey());
+ ServerName existingServer = tableKeys.get(startKey);
+ if (!sn.equals(existingServer)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("There is a mismatch in the existing server name for the region " + hri
+ + ". Replacing the server " + existingServer + " with " + sn + ".");
+ }
+ tableKeys.put(startKey, sn);
+ }
+ }
+
+ /**
+ * Prepare region plans for cluster state
+ * @param clusterState if balancing is table wise then cluster state contains only indexed or
+ * index table regions, otherwise it contains all index tables regions.
+ * @param regionPlans
+ * @return
+ */
+ private List<RegionPlan> prepareRegionPlansForClusterState(
+ Map<ServerName, List<HRegionInfo>> clusterState, List<RegionPlan> regionPlans) {
+ if (regionPlans == null) regionPlans = new ArrayList<RegionPlan>();
+ ImmutableBytesWritable startKey = new ImmutableBytesWritable();
+ for (Entry<ServerName, List<HRegionInfo>> serverVsRegionList : clusterState.entrySet()) {
+ List<HRegionInfo> regionInfos = serverVsRegionList.getValue();
+ ServerName server = serverVsRegionList.getKey();
+ for (HRegionInfo regionInfo : regionInfos) {
+ if (!isTableColocated(regionInfo.getTable())) continue;
+ TableName mappedTableName = getMappedTableToColocate(regionInfo.getTable());
+ startKey.set(regionInfo.getStartKey());
+ ServerName sn = this.colocationInfo.get(mappedTableName).get(startKey);
+ if (sn.equals(server)) {
+ continue;
+ } else {
+ RegionPlan rp = new RegionPlan(regionInfo, server, sn);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Selected server " + rp.getDestination()
+ + " as destination for region "
+ + regionInfo.getRegionNameAsString() + " from colocation info.");
+ }
+ regionOnline(regionInfo, rp.getDestination());
+ regionPlans.add(rp);
+ }
+ }
+ }
+ return regionPlans;
+ }
+
+ private void updateRegionPlans(List<RegionPlan> regionPlans) {
+ for (RegionPlan regionPlan : regionPlans) {
+ HRegionInfo hri = regionPlan.getRegionInfo();
+ if (!isTableColocated(hri.getTable())) continue;
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Saving region plan of region " + hri.getRegionNameAsString() + '.');
+ }
+ regionOnline(hri, regionPlan.getDestination());
+ }
+ }
+
+ @Override
+ public Map<ServerName, List<HRegionInfo>> roundRobinAssignment(List<HRegionInfo> regions,
+ List<ServerName> servers) throws HBaseIOException {
+ List<HRegionInfo> userRegions = new ArrayList<HRegionInfo>();
+ List<HRegionInfo> indexRegions = new ArrayList<HRegionInfo>();
+ for (HRegionInfo hri : regions) {
+ seperateUserAndIndexRegion(hri, userRegions, indexRegions);
+ }
+ Map<ServerName, List<HRegionInfo>> bulkPlan = null;
+ if (!userRegions.isEmpty()) {
+ bulkPlan = this.delegator.roundRobinAssignment(userRegions, servers);
+ // This should not happen.
+ if (null == bulkPlan) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No region plans selected for user regions in roundRobinAssignment.");
+ }
+ return null;
+ }
+ savePlan(bulkPlan);
+ }
+ bulkPlan = prepareIndexRegionsPlan(indexRegions, bulkPlan, servers);
+ return bulkPlan;
+ }
+
+ private void seperateUserAndIndexRegion(HRegionInfo hri, List<HRegionInfo> userRegions,
+ List<HRegionInfo> indexRegions) {
+ if (indexTableVsUserTable.containsKey(hri.getTable())) {
+ indexRegions.add(hri);
+ return;
+ }
+ userRegions.add(hri);
+ }
+
+ private Map<ServerName, List<HRegionInfo>> prepareIndexRegionsPlan(
+ List<HRegionInfo> indexRegions, Map<ServerName, List<HRegionInfo>> bulkPlan,
+ List<ServerName> servers) throws HBaseIOException {
+ if (null != indexRegions && !indexRegions.isEmpty()) {
+ if (null == bulkPlan) {
+ bulkPlan = new ConcurrentHashMap<ServerName, List<HRegionInfo>>();
+ }
+ for (HRegionInfo hri : indexRegions) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Preparing region plan for index region "
+ + hri.getRegionNameAsString() + '.');
+ }
+ ServerName destServer = getDestServerForIdxRegion(hri);
+ List<HRegionInfo> destServerRegions = null;
+ if (destServer == null) destServer = this.randomAssignment(hri, servers);
+ if (destServer != null) {
+ destServerRegions = bulkPlan.get(destServer);
+ if (null == destServerRegions) {
+ destServerRegions = new ArrayList<HRegionInfo>();
+ bulkPlan.put(destServer, destServerRegions);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Server " + destServer + " selected for region "
+ + hri.getRegionNameAsString() + '.');
+ }
+ destServerRegions.add(hri);
+ regionOnline(hri, destServer);
+ }
+ }
+ }
+ return bulkPlan;
+ }
+
+ private ServerName getDestServerForIdxRegion(HRegionInfo hri) {
+ // Every time we calculate the table name because in case of master restart the index
+ // regions
+ // may be coming for different index tables.
+ TableName actualTable = getMappedTableToColocate(hri.getTable());
+ ImmutableBytesWritable startkey = new ImmutableBytesWritable(hri.getStartKey());
+ synchronized (this.colocationInfo) {
+
+ Map<ImmutableBytesWritable, ServerName> tableKeys = colocationInfo.get(actualTable);
+ if (null == tableKeys) {
+ // Can this case come
+ return null;
+ }
+ if (tableKeys.containsKey(startkey)) {
+ // put index region location if corresponding user region found in regionLocation
+ // map.
+ ServerName sn = tableKeys.get(startkey);
+ regionOnline(hri, sn);
+ return sn;
+ }
+ }
+ return null;
+ }
+
+ private void savePlan(Map<ServerName, List<HRegionInfo>> bulkPlan) {
+ synchronized (this.colocationInfo) {
+ for (Entry<ServerName, List<HRegionInfo>> e : bulkPlan.entrySet()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Saving user regions' plans for server " + e.getKey() + '.');
+ }
+ for (HRegionInfo hri : e.getValue()) {
+ if (!isTableColocated(hri.getTable())) continue;
+ regionOnline(hri, e.getKey());
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Saved user regions' plans for server " + e.getKey() + '.');
+ }
+ }
+ }
+ }
+
+ @Override
+ public Map<ServerName, List<HRegionInfo>> retainAssignment(
+ Map<HRegionInfo, ServerName> regions, List<ServerName> servers) throws HBaseIOException {
+ Map<HRegionInfo, ServerName> userRegionsMap =
+ new ConcurrentHashMap<HRegionInfo, ServerName>();
+ List<HRegionInfo> indexRegions = new ArrayList<HRegionInfo>();
+ for (Entry<HRegionInfo, ServerName> e : regions.entrySet()) {
+ seperateUserAndIndexRegion(e, userRegionsMap, indexRegions, servers);
+ }
+ Map<ServerName, List<HRegionInfo>> bulkPlan = null;
+ if (!userRegionsMap.isEmpty()) {
+ bulkPlan = this.delegator.retainAssignment(userRegionsMap, servers);
+ if (bulkPlan == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Empty region plan for user regions.");
+ }
+ return null;
+ }
+ savePlan(bulkPlan);
+ }
+ bulkPlan = prepareIndexRegionsPlan(indexRegions, bulkPlan, servers);
+ return bulkPlan;
+ }
+
+ private void seperateUserAndIndexRegion(Entry<HRegionInfo, ServerName> e,
+ Map<HRegionInfo, ServerName> userRegionsMap, List<HRegionInfo> indexRegions,
+ List<ServerName> servers) {
+ HRegionInfo hri = e.getKey();
+ if (indexTableVsUserTable.containsKey(hri.getTable())) {
+ indexRegions.add(hri);
+ return;
+ }
+ if (e.getValue() == null) {
+ userRegionsMap.put(hri, servers.get(RANDOM.nextInt(servers.size())));
+ } else {
+ userRegionsMap.put(hri, e.getValue());
+ }
+ }
+
+ @Override
+ public Map<HRegionInfo, ServerName> immediateAssignment(List<HRegionInfo> regions,
+ List<ServerName> servers) throws HBaseIOException {
+ return this.delegator.immediateAssignment(regions, servers);
+ }
+
+ @Override
+ public ServerName randomAssignment(HRegionInfo regionInfo, List<ServerName> servers)
+ throws HBaseIOException {
+ if (!isTableColocated(regionInfo.getTable())) {
+ return this.delegator.randomAssignment(regionInfo, servers);
+ }
+ ServerName sn = getServerNameFromMap(regionInfo, servers);
+ if (sn == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No server found for region " + regionInfo.getRegionNameAsString() + '.');
+ }
+ sn = getRandomServer(regionInfo, servers);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Destination server for region " + regionInfo.getRegionNameAsString()
+ + " is " + ((sn == null) ? "null" : sn.toString()) + '.');
+ }
+ return sn;
+ }
+
+ private ServerName getRandomServer(HRegionInfo regionInfo, List<ServerName> servers)
+ throws HBaseIOException {
+ ServerName sn = null;
+ sn = this.delegator.randomAssignment(regionInfo, servers);
+ if (sn == null) return null;
+ regionOnline(regionInfo, sn);
+ return sn;
+ }
+
+ private ServerName getServerNameFromMap(HRegionInfo regionInfo, List<ServerName> onlineServers) {
+ TableName tableName = regionInfo.getTable();
+ TableName mappedTable = getMappedTableToColocate(regionInfo.getTable());
+ ImmutableBytesWritable startKey = new ImmutableBytesWritable(regionInfo.getStartKey());
+ synchronized (this.colocationInfo) {
+ Map<ImmutableBytesWritable, ServerName> correspondingTableKeys =
+ this.colocationInfo.get(mappedTable);
+ Map<ImmutableBytesWritable, ServerName> actualTableKeys =
+ this.colocationInfo.get(tableName);
+
+ if (null != correspondingTableKeys) {
+ if (correspondingTableKeys.containsKey(startKey)) {
+ ServerName previousServer = null;
+ if (null != actualTableKeys) {
+ previousServer = actualTableKeys.get(startKey);
+ }
+ ServerName sn = correspondingTableKeys.get(startKey);
+ if (null != previousServer) {
+ // if servername of index region and user region are same in colocationInfo
+ // clean
+ // previous plans and return null
+ if (previousServer.equals(sn)) {
+ correspondingTableKeys.remove(startKey);
+ actualTableKeys.remove(startKey);
+ if (LOG.isDebugEnabled()) {
+ LOG
+ .debug("Both user region plan and corresponding index region plan "
+ + "in colocation info are same. Hence clearing the plans to select new plan"
+ + " for the region "
+ + regionInfo.getRegionNameAsString() + ".");
+ }
+ return null;
+ }
+ }
+ if (sn != null && onlineServers.contains(sn)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Updating the region plan of the region "
+ + regionInfo.getRegionNameAsString() + " with server " + sn);
+ }
+ regionOnline(regionInfo, sn);
+ return sn;
+ } else if (sn != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("The location " + sn + " of region with start key"
+ + Bytes.toStringBinary(regionInfo.getStartKey())
+ + " is not in online. Selecting other region server.");
+ }
+ return null;
+ }
+ }
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No region plans in colocationInfo for table " + mappedTable);
+ }
+ }
+ return null;
+ }
+ }
+
+ @Override
+ public void regionOnline(HRegionInfo regionInfo, ServerName sn) {
+ TableName tableName = regionInfo.getTable();
+ synchronized (this.colocationInfo) {
+ Map<ImmutableBytesWritable, ServerName> tabkeKeys = this.colocationInfo.get(tableName);
+ if (tabkeKeys == null) {
+ tabkeKeys = new ConcurrentHashMap<ImmutableBytesWritable, ServerName>();
+ this.colocationInfo.put(tableName, tabkeKeys);
+ }
+ tabkeKeys.put(new ImmutableBytesWritable(regionInfo.getStartKey()), sn);
+ }
+ }
+
+ public void clearTableRegionPlans(TableName tableName) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Clearing regions plans from colocationInfo for table " + tableName);
+ }
+ synchronized (this.colocationInfo) {
+ this.colocationInfo.remove(tableName);
+ }
+ }
+
+ @Override
+ public void regionOffline(HRegionInfo regionInfo) {
+ TableName tableName = regionInfo.getTable();
+ synchronized (this.colocationInfo) {
+ Map<ImmutableBytesWritable, ServerName> tableKeys = this.colocationInfo.get(tableName);
+ if (null == tableKeys) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No regions of table " + tableName + " in the colocationInfo.");
+ }
+ } else {
+ tableKeys.remove(new ImmutableBytesWritable(regionInfo.getStartKey()));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("The regioninfo " + regionInfo + " removed from the colocationInfo");
+ }
+ }
+ }
+ }
+
+ @Override
+ public boolean isStopped() {
+ return stopped;
+ }
+
+ @Override
+ public void stop(String why) {
+ LOG.info("Load Balancer stop requested: " + why);
+ stopped = true;
+ }
+
+ public void populateTablesToColocate(Map<String, HTableDescriptor> tableDescriptors) {
+ HTableDescriptor desc = null;
+ for (Entry<String, HTableDescriptor> entry : tableDescriptors.entrySet()) {
+ desc = entry.getValue();
+ if (desc.getValue(PARENT_TABLE_KEY) != null) {
+ addTablesToColocate(TableName.valueOf(desc.getValue(PARENT_TABLE_KEY)), desc
+ .getTableName());
+ }
+ }
+ }
+
+ /**
+ * Add tables whose regions to co-locate.
+ * @param userTable
+ * @param indexTable
+ */
+ public void addTablesToColocate(TableName userTable, TableName indexTable) {
+ if (userTable.equals(indexTable)) {
+ throw new IllegalArgumentException("Tables to colocate should not be same.");
+ } else if (isTableColocated(userTable)) {
+ throw new IllegalArgumentException("User table already colocated with table "
+ + getMappedTableToColocate(userTable));
+ } else if (isTableColocated(indexTable)) {
+ throw new IllegalArgumentException("Index table is already colocated with table "
+ + getMappedTableToColocate(indexTable));
+ }
+ userTableVsIndexTable.put(userTable, indexTable);
+ indexTableVsUserTable.put(indexTable, userTable);
+ }
+
+ /**
+ * Removes the specified table and corresponding table from co-location.
+ * @param table
+ */
+ public void removeTablesFromColocation(TableName table) {
+ TableName other = userTableVsIndexTable.remove(table);
+ if (other != null) {
+ indexTableVsUserTable.remove(other);
+ } else {
+ other = indexTableVsUserTable.remove(table);
+ if (other != null) userTableVsIndexTable.remove(table);
+ }
+ }
+
+ /**
+ * Return mapped table to co-locate.
+ * @param tableName
+ * @return index table if the specified table is user table or vice versa.
+ */
+ public TableName getMappedTableToColocate(TableName tableName) {
+ TableName other = userTableVsIndexTable.get(tableName);
+ return other == null ? indexTableVsUserTable.get(tableName) : other;
+ }
+
+ public boolean isTableColocated(TableName table) {
+ return userTableVsIndexTable.containsKey(table) || indexTableVsUserTable.containsKey(table);
+ }
+
+ /**
+ * Populates table's region locations into co-location info from master.
+ * @param table
+ */
+ public void populateRegionLocations(TableName table) {
+ synchronized (this.colocationInfo) {
+ if (!isTableColocated(table)) {
+ throw new IllegalArgumentException("Specified table " + table
+ + " should be in one of the tables to co-locate.");
+ }
+ RegionStates regionStates = this.master.getAssignmentManager().getRegionStates();
+ List<HRegionInfo> onlineRegions = regionStates.getRegionsOfTable(table);
+ for (HRegionInfo hri : onlineRegions) {
+ regionOnline(hri, regionStates.getRegionServerOfRegion(hri));
+ }
+ Map<String, RegionState> regionsInTransition = regionStates.getRegionsInTransition();
+ for (RegionState regionState : regionsInTransition.values()) {
+ if (table.equals(regionState.getRegion().getTable())
+ && regionState.getServerName() != null) {
+ regionOnline(regionState.getRegion(), regionState.getServerName());
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/master/IndexMasterObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/master/IndexMasterObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/master/IndexMasterObserver.java
new file mode 100644
index 0000000..dfb2c62
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/master/IndexMasterObserver.java
@@ -0,0 +1,69 @@
+/*
+ * 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.hbase.index.master;
+
+import java.io.IOException;
+
+import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.coprocessor.BaseMasterObserver;
+import org.apache.hadoop.hbase.coprocessor.MasterCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.ObserverContext;
+import org.apache.hadoop.hbase.master.LoadBalancer;
+import org.apache.phoenix.hbase.index.balancer.IndexLoadBalancer;
+
+/**
+ * Defines of coprocessor hooks(to support secondary indexing) of operations on
+ * {@link org.apache.hadoop.hbase.master.HMaster} process.
+ */
+public class IndexMasterObserver extends BaseMasterObserver {
+ IndexLoadBalancer balancer = null;
+
+ @Override
+ public void preMasterInitialization(ObserverContext<MasterCoprocessorEnvironment> ctx)
+ throws IOException {
+ LoadBalancer loadBalancer =
+ ctx.getEnvironment().getMasterServices().getAssignmentManager().getBalancer();
+ if (loadBalancer instanceof IndexLoadBalancer) {
+ balancer = (IndexLoadBalancer) loadBalancer;
+ }
+ super.preMasterInitialization(ctx);
+ }
+
+ @Override
+ public void preCreateTableHandler(ObserverContext<MasterCoprocessorEnvironment> ctx,
+ HTableDescriptor desc, HRegionInfo[] regions) throws IOException {
+ TableName userTableName = null;
+ if (balancer != null && desc.getValue(IndexLoadBalancer.PARENT_TABLE_KEY) != null) {
+ userTableName =
+ TableName.valueOf(desc.getValue(IndexLoadBalancer.PARENT_TABLE_KEY));
+ balancer.addTablesToColocate(userTableName, desc.getTableName());
+ }
+ if (userTableName != null) balancer.populateRegionLocations(userTableName);
+ super.preCreateTableHandler(ctx, desc, regions);
+ }
+
+ @Override
+ public void postDeleteTableHandler(ObserverContext<MasterCoprocessorEnvironment> ctx,
+ TableName tableName) throws IOException {
+ if (balancer != null && balancer.isTableColocated(tableName)) {
+ balancer.removeTablesFromColocation(tableName);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/9a5529c8/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
index 4cbf6dd..db38515 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
@@ -120,11 +120,11 @@ public class IndexManagementUtil {
};
}
- private static class ReferencingColumn {
+ public static class ReferencingColumn {
ImmutableBytesPtr family;
ImmutableBytesPtr qual;
- static ReferencingColumn wrap(ColumnReference ref) {
+ public static ReferencingColumn wrap(ColumnReference ref) {
ImmutableBytesPtr family = new ImmutableBytesPtr(ref.getFamily());
ImmutableBytesPtr qual = new ImmutableBytesPtr(ref.getQualifier());
return new ReferencingColumn(family, qual);