You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/01/27 23:15:50 UTC
[34/51] [partial] Initial commit of master branch from github
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/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
new file mode 100644
index 0000000..677328e
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionCode.java
@@ -0,0 +1,340 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.exception;
+
+import java.sql.SQLException;
+import java.util.Map;
+
+import org.apache.hadoop.hbase.index.util.IndexManagementUtil;
+import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.schema.AmbiguousColumnException;
+import org.apache.phoenix.schema.AmbiguousTableException;
+import org.apache.phoenix.schema.ColumnAlreadyExistsException;
+import org.apache.phoenix.schema.ColumnFamilyNotFoundException;
+import org.apache.phoenix.schema.ColumnNotFoundException;
+import org.apache.phoenix.schema.ConcurrentTableMutationException;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.ReadOnlyTableException;
+import org.apache.phoenix.schema.SequenceAlreadyExistsException;
+import org.apache.phoenix.schema.SequenceNotFoundException;
+import org.apache.phoenix.schema.TableAlreadyExistsException;
+import org.apache.phoenix.schema.TableNotFoundException;
+import org.apache.phoenix.schema.TypeMismatchException;
+import org.apache.phoenix.util.MetaDataUtil;
+
+import com.google.common.collect.Maps;
+
+
+/**
+ * Various SQLException Information. Including a vender-specific errorcode and a standard SQLState.
+ *
+ * @author zhuang
+ * @since 1.0
+ */
+public enum SQLExceptionCode {
+
+ /**
+ * Connection Exception (errorcode 01, sqlstate 08)
+ */
+ IO_EXCEPTION(101, "08000", "Unexpected IO exception."),
+ MALFORMED_CONNECTION_URL(102, "08001", "Malformed connection url."),
+ CANNOT_ESTABLISH_CONNECTION(103, "08004", "Unable to establish connection."),
+
+ /**
+ * Data Exception (errorcode 02, sqlstate 22)
+ */
+ ILLEGAL_DATA(201, "22000", "Illegal data."),
+ DIVIDE_BY_ZERO(202, "22012", "Divide by zero."),
+ TYPE_MISMATCH(203, "22005", "Type mismatch.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new TypeMismatchException(info.getMessage());
+ }
+ }),
+ VALUE_IN_UPSERT_NOT_CONSTANT(204, "22008", "Values in UPSERT must evaluate to a constant."),
+ MALFORMED_URL(205, "22009", "Malformed URL."),
+ DATA_INCOMPATIBLE_WITH_TYPE(206, "22003", "The value is outside the range for the data type."),
+ MISSING_CHAR_LENGTH(207, "22003", "Missing length for CHAR."),
+ NONPOSITIVE_CHAR_LENGTH(208, "22003", "CHAR or VARCHAR must have a positive length."),
+ DECIMAL_PRECISION_OUT_OF_RANGE(209, "22003", "Decimal precision outside of range. Should be within 1 and " + PDataType.MAX_PRECISION + "."),
+ MISSING_BINARY_LENGTH(210, "22003", "Missing length for BINARY."),
+ NONPOSITIVE_BINARY_LENGTH(211, "22003", "BINARY must have a positive length."),
+ SERVER_ARITHMETIC_ERROR(212, "22012", "Arithmetic error on server."),
+ VALUE_OUTSIDE_RANGE(213,"22003","Value outside range."),
+ VALUE_IN_LIST_NOT_CONSTANT(214, "22008", "Values in IN must evaluate to a constant."),
+
+ /**
+ * Constraint Violation (errorcode 03, sqlstate 23)
+ */
+ CONCURRENT_TABLE_MUTATION(301, "23000", "Concurrent modification to table.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new ConcurrentTableMutationException(info.getSchemaName(), info.getTableName());
+ }
+ }),
+ CANNOT_INDEX_COLUMN_ON_TYPE(302, "23100", "The column cannot be index due to its type."),
+
+ /**
+ * Invalid Cursor State (errorcode 04, sqlstate 24)
+ */
+ CURSOR_BEFORE_FIRST_ROW(401, "24015","Cursor before first row."),
+ CURSOR_PAST_LAST_ROW(402, "24016", "Cursor past last row."),
+
+ /**
+ * Syntax Error or Access Rule Violation (errorcode 05, sqlstate 42)
+ */
+ AMBIGUOUS_TABLE(501, "42000", "Table name exists in more than one table schema and is used without being qualified.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new AmbiguousTableException(info.getTableName(), info.getRootCause());
+ }
+ }),
+ AMBIGUOUS_COLUMN(502, "42702", "Column reference ambiguous or duplicate names.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new AmbiguousColumnException(info.getColumnName(), info.getRootCause());
+ }
+ }),
+ INDEX_MISSING_PK_COLUMNS(503, "42602", "Index table missing PK Columns."),
+ COLUMN_NOT_FOUND(504, "42703", "Undefined column.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new ColumnNotFoundException(info.getSchemaName(), info.getTableName(), info.getFamilyName(), info.getColumnName());
+ }
+ }),
+ READ_ONLY_TABLE(505, "42000", "Table is read only.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new ReadOnlyTableException(info.getMessage(), info.getSchemaName(), info.getTableName(), info.getFamilyName());
+ }
+ }),
+ CANNOT_DROP_PK(506, "42817", "Primary key column may not be dropped."),
+ CANNOT_CONVERT_TYPE(507, "42846", "Cannot convert type."),
+ PRIMARY_KEY_MISSING(509, "42888", "The table does not have a primary key."),
+ PRIMARY_KEY_ALREADY_EXISTS(510, "42889", "The table already has a primary key."),
+ ORDER_BY_NOT_IN_SELECT_DISTINCT(511, "42890", "All ORDER BY expressions must appear in SELECT DISTINCT:"),
+ INVALID_PRIMARY_KEY_CONSTRAINT(512, "42891", "Invalid column reference in primary key constraint"),
+ ARRAY_NOT_ALLOWED_IN_PRIMARY_KEY(513, "42892", "Array type not allowed as primary key constraint"),
+ COLUMN_EXIST_IN_DEF(514, "42892", "A duplicate column name was detected in the object definition or ALTER TABLE statement.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new ColumnAlreadyExistsException(info.getSchemaName(), info.getTableName(), info.getColumnName());
+ }
+ }),
+ ORDER_BY_ARRAY_NOT_SUPPORTED(515, "42893", "ORDER BY of an array type is not allowed"),
+ NON_EQUALITY_ARRAY_COMPARISON(516, "42894", "Array types may only be compared using = or !="),
+
+ /**
+ * HBase and Phoenix specific implementation defined sub-classes.
+ * Column family related exceptions.
+ *
+ * For the following exceptions, use errorcode 10.
+ */
+ COLUMN_FAMILY_NOT_FOUND(1001, "42I01", "Undefined column family.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new ColumnFamilyNotFoundException(info.getFamilyName());
+ }
+ }),
+ PROPERTIES_FOR_FAMILY(1002, "42I02","Properties may not be defined for an unused family name."),
+ // Primary/row key related exceptions.
+ PRIMARY_KEY_WITH_FAMILY_NAME(1003, "42J01", "Primary key columns must not have a family name."),
+ PRIMARY_KEY_OUT_OF_ORDER(1004, "42J02", "Order of columns in primary key constraint must match the order in which they're declared."),
+ VARBINARY_IN_ROW_KEY(1005, "42J03", "The VARBINARY type can only be used as the last part of a multi-part row key."),
+ NOT_NULLABLE_COLUMN_IN_ROW_KEY(1006, "42J04", "Only nullable columns may be added to a multi-part row key."),
+ VARBINARY_LAST_PK(1015, "42J04", "Cannot add column to table when the last PK column is of type VARBINARY."),
+ NULLABLE_FIXED_WIDTH_LAST_PK(1023, "42J04", "Cannot add column to table when the last PK column is nullable and fixed width."),
+ CANNOT_MODIFY_VIEW_PK(1036, "42J04", "Cannot modify the primary key of a VIEW."),
+ BASE_TABLE_COLUMN(1037, "42J04", "Cannot modify columns of base table used by tenant-specific tables."),
+ // Key/value column related errors
+ KEY_VALUE_NOT_NULL(1007, "42K01", "A key/value column may not be declared as not null."),
+ // View related errors.
+ VIEW_WITH_TABLE_CONFIG(1008, "42L01", "A view may not contain table configuration properties."),
+ VIEW_WITH_PROPERTIES(1009, "42L02", "Properties may not be defined for a view."),
+ // Table related errors that are not in standard code.
+ CANNOT_MUTATE_TABLE(1010, "42M01", "Not allowed to mutate table."),
+ UNEXPECTED_MUTATION_CODE(1011, "42M02", "Unexpected mutation code."),
+ TABLE_UNDEFINED(1012, "42M03", "Table undefined.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new TableNotFoundException(info.getSchemaName(), info.getTableName());
+ }
+ }),
+ TABLE_ALREADY_EXIST(1013, "42M04", "Table already exists.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new TableAlreadyExistsException(info.getSchemaName(), info.getTableName());
+ }
+ }),
+
+ // Syntax error
+ TYPE_NOT_SUPPORTED_FOR_OPERATOR(1014, "42Y01", "The operator does not support the operand type."),
+ AGGREGATE_IN_GROUP_BY(1016, "42Y26", "Aggregate expressions may not be used in GROUP BY."),
+ AGGREGATE_IN_WHERE(1017, "42Y26", "Aggregate may not be used in WHERE."),
+ AGGREGATE_WITH_NOT_GROUP_BY_COLUMN(1018, "42Y27", "Aggregate may not contain columns not in GROUP BY."),
+ ONLY_AGGREGATE_IN_HAVING_CLAUSE(1019, "42Y26", "Only aggregate maybe used in the HAVING clause."),
+ UPSERT_COLUMN_NUMBERS_MISMATCH(1020, "42Y60", "Number of columns upserting must match number of values."),
+ // Table properties exception.
+ INVALID_BUCKET_NUM(1021, "42Y80", "Salt bucket numbers should be with 1 and 256."),
+ NO_SPLITS_ON_SALTED_TABLE(1022, "42Y81", "Should not specify split points on salted table with default row key order."),
+ SALT_ONLY_ON_CREATE_TABLE(1024, "42Y83", "Salt bucket number may only be specified when creating a table."),
+ SET_UNSUPPORTED_PROP_ON_ALTER_TABLE(1025, "42Y84", "Unsupported property set in ALTER TABLE command."),
+ CANNOT_ADD_NOT_NULLABLE_COLUMN(1038, "42Y84", "Only nullable columns may be added for a pre-existing table."),
+ NO_MUTABLE_INDEXES(1026, "42Y85", "Mutable secondary indexes are only supported for HBase version " + MetaDataUtil.decodeHBaseVersionAsString(PhoenixDatabaseMetaData.MUTABLE_SI_VERSION_THRESHOLD) + " and above."),
+ NO_DELETE_IF_IMMUTABLE_INDEX(1027, "42Y86", "Delete not allowed on a table with IMMUTABLE_ROW with non PK column in index."),
+ INVALID_INDEX_STATE_TRANSITION(1028, "42Y87", "Invalid index state transition."),
+ INVALID_MUTABLE_INDEX_CONFIG(1029, "42Y88", "Mutable secondary indexes must have the "
+ + IndexManagementUtil.WAL_EDIT_CODEC_CLASS_KEY + " property set to "
+ + IndexManagementUtil.INDEX_WAL_EDIT_CODEC_CLASS_NAME + " in the hbase-sites.xml of every region server"),
+
+
+ CANNOT_CREATE_TENANT_SPECIFIC_TABLE(1030, "42Y89", "Cannot create table for tenant-specific connection"),
+ CANNOT_DEFINE_PK_FOR_VIEW(1031, "42Y90", "Defining PK columns for a VIEW is not allowed."),
+ DEFAULT_COLUMN_FAMILY_ONLY_ON_CREATE_TABLE(1034, "42Y93", "Default column family may only be specified when creating a table."),
+ INSUFFICIENT_MULTI_TENANT_COLUMNS(1040, "42Y96", "A MULTI_TENANT table must have 2 or more PK columns with the first column being 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 used in WHERE clause of VIEW may not be updated."),
+
+ /** Sequence related */
+ SEQUENCE_ALREADY_EXIST(1200, "42Z00", "Sequence already exists.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new SequenceAlreadyExistsException(info.getSchemaName(), info.getTableName());
+ }
+ }),
+ SEQUENCE_UNDEFINED(1201, "42Z01", "Sequence undefined.", new Factory() {
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new SequenceNotFoundException(info.getSchemaName(), info.getTableName());
+ }
+ }),
+ STARTS_WITH_MUST_BE_CONSTANT(1202, "42Z02", "Sequence STARTS WITH value must be an integer or long constant."),
+ INCREMENT_BY_MUST_BE_CONSTANT(1203, "42Z03", "Sequence INCREMENT BY value must be an integer or long constant."),
+ CACHE_MUST_BE_NON_NEGATIVE_CONSTANT(1204, "42Z04", "Sequence CACHE value must be a non negative integer constant."),
+ INVALID_USE_OF_NEXT_VALUE_FOR(1205, "42Z05", "NEXT VALUE FOR may only be used as in a SELECT or an UPSERT VALUES expression."),
+ CANNOT_CALL_CURRENT_BEFORE_NEXT_VALUE(1206, "42Z06", "NEXT VALUE FOR must be called before CURRENT VALUE FOR is called."),
+ EMPTY_SEQUENCE_CACHE(1207, "42Z07", "No more cached sequence values"),
+
+ /** Parser error. (errorcode 06, sqlState 42P) */
+ PARSER_ERROR(601, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
+ MISSING_TOKEN(602, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
+ UNWANTED_TOKEN(603, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
+ MISMATCHED_TOKEN(604, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
+ UNKNOWN_FUNCTION(605, "42P00", "Syntax error.", Factory.SYTAX_ERROR),
+
+ /**
+ * Implementation defined class. Execution exceptions (errorcode 11, sqlstate XCL).
+ */
+ RESULTSET_CLOSED(1101, "XCL01", "ResultSet is closed."),
+ GET_TABLE_REGIONS_FAIL(1102, "XCL02", "Cannot get all table regions"),
+ EXECUTE_QUERY_NOT_APPLICABLE(1103, "XCL03", "executeQuery may not be used."),
+ EXECUTE_UPDATE_NOT_APPLICABLE(1104, "XCL03", "executeUpdate may not be used."),
+ SPLIT_POINT_NOT_CONSTANT(1105, "XCL04", "Split points must be constants."),
+
+ /**
+ * Implementation defined class. Phoenix internal error. (errorcode 20, sqlstate INT).
+ */
+ CANNOT_CALL_METHOD_ON_TYPE(2001, "INT01", "Cannot call method on the argument type."),
+ CLASS_NOT_UNWRAPPABLE(2002, "INT03", "Class not unwrappable"),
+ PARAM_INDEX_OUT_OF_BOUND(2003, "INT04", "Parameter position is out of range."),
+ PARAM_VALUE_UNBOUND(2004, "INT05", "Parameter value unbound"),
+ INTERRUPTED_EXCEPTION(2005, "INT07", "Interrupted exception."),
+ INCOMPATIBLE_CLIENT_SERVER_JAR(2006, "INT08", "Incompatible jars detected between client and server."),
+ OUTDATED_JARS(2007, "INT09", "Outdated jars."),
+ INDEX_METADATA_NOT_FOUND(2008, "INT10", "Unable to find cached index metadata. "),
+ UNKNOWN_ERROR_CODE(2009, "INT11", "Unknown error code"),
+ ;
+
+ private final int errorCode;
+ private final String sqlState;
+ private final String message;
+ private final Factory factory;
+
+ private SQLExceptionCode(int errorCode, String sqlState, String message) {
+ this(errorCode, sqlState, message, Factory.DEFAULTY);
+ }
+
+ private SQLExceptionCode(int errorCode, String sqlState, String message, Factory factory) {
+ this.errorCode = errorCode;
+ this.sqlState = sqlState;
+ this.message = message;
+ this.factory = factory;
+ }
+
+ public String getSQLState() {
+ return sqlState;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public int getErrorCode() {
+ return errorCode;
+ }
+
+ @Override
+ public String toString() {
+ return "ERROR " + errorCode + " (" + sqlState + "): " + message;
+ }
+
+ public Factory getExceptionFactory() {
+ return factory;
+ }
+
+ public static interface Factory {
+ public static final Factory DEFAULTY = new Factory() {
+
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new SQLException(info.toString(), info.getCode().getSQLState(), info.getCode().getErrorCode(), info.getRootCause());
+ }
+
+ };
+ public static final Factory SYTAX_ERROR = new Factory() {
+
+ @Override
+ public SQLException newException(SQLExceptionInfo info) {
+ return new PhoenixParserException(info.getMessage(), info.getRootCause());
+ }
+
+ };
+ public SQLException newException(SQLExceptionInfo info);
+ }
+
+ private static final Map<Integer,SQLExceptionCode> errorCodeMap = Maps.newHashMapWithExpectedSize(SQLExceptionCode.values().length);
+ static {
+ for (SQLExceptionCode code : SQLExceptionCode.values()) {
+ SQLExceptionCode otherCode = errorCodeMap.put(code.getErrorCode(), code);
+ if (otherCode != null) {
+ throw new IllegalStateException("Duplicate error code for " + code + " and " + otherCode);
+ }
+ }
+ }
+
+ public static SQLExceptionCode fromErrorCode(int errorCode) throws SQLException {
+ SQLExceptionCode code = errorCodeMap.get(errorCode);
+ if (code == null) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.UNKNOWN_ERROR_CODE)
+ .setMessage(Integer.toString(errorCode)).build().buildException();
+ }
+ return code;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionInfo.java
new file mode 100644
index 0000000..e025054
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/SQLExceptionInfo.java
@@ -0,0 +1,166 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.exception;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.util.SchemaUtil;
+
+
+/**
+ * Object serves as a closure of all coordinate information for SQLException messages.
+ *
+ * @author zhuang
+ * @since 1.0
+ */
+public class SQLExceptionInfo {
+
+ /**
+ * Constants used in naming exception location.
+ */
+ public static final String SCHEMA_NAME = "schemaName";
+ public static final String TABLE_NAME = "tableName";
+ public static final String FAMILY_NAME = "familyName";
+ public static final String COLUMN_NAME = "columnName";
+
+ private final Throwable rootCause;
+ private final SQLExceptionCode code; // Should always have one.
+ private final String message;
+ private final String schemaName;
+ private final String tableName;
+ private final String familyName;
+ private final String columnName;
+
+ public static class Builder {
+
+ private Throwable rootCause;
+ private SQLExceptionCode code; // Should always have one.
+ private String message;
+ private String schemaName;
+ private String tableName;
+ private String familyName;
+ private String columnName;
+
+ public Builder(SQLExceptionCode code) {
+ this.code = code;
+ }
+
+ public Builder setRootCause(Throwable t) {
+ this.rootCause = t;
+ return this;
+ }
+
+ public Builder setMessage(String message) {
+ this.message = message;
+ return this;
+ }
+
+ public Builder setSchemaName(String schemaName) {
+ this.schemaName = schemaName;
+ return this;
+ }
+
+ public Builder setTableName(String tableName) {
+ this.tableName = tableName;
+ return this;
+ }
+
+ public Builder setFamilyName(String familyName) {
+ this.familyName = familyName;
+ return this;
+ }
+
+ public Builder setColumnName(String columnName) {
+ this.columnName = columnName;
+ return this;
+ }
+
+ public SQLExceptionInfo build() {
+ return new SQLExceptionInfo(this);
+ }
+
+ @Override
+ public String toString() {
+ return code.toString();
+ }
+ }
+
+ private SQLExceptionInfo(Builder builder) {
+ code = builder.code;
+ rootCause = builder.rootCause;
+ message = builder.message;
+ schemaName = builder.schemaName;
+ tableName = builder.tableName;
+ familyName = builder.familyName;
+ columnName = builder.columnName;
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder(code.toString());
+ if (message != null) {
+ builder.append(" ").append(message);
+ }
+ String columnDisplayName = SchemaUtil.getMetaDataEntityName(schemaName, tableName, familyName, columnName);
+ if (columnName != null) {
+ builder.append(" ").append(COLUMN_NAME).append("=").append(columnDisplayName);
+ } else if (familyName != null) {
+ builder.append(" ").append(FAMILY_NAME).append("=").append(columnDisplayName);
+ } else if (tableName != null) {
+ builder.append(" ").append(TABLE_NAME).append("=").append(columnDisplayName);
+ } else if (schemaName != null) {
+ builder.append(" ").append(SCHEMA_NAME).append("=").append(columnDisplayName);
+ }
+ return builder.toString();
+ }
+
+ public SQLException buildException() {
+ return code.getExceptionFactory().newException(this);
+ }
+
+ public Throwable getRootCause() {
+ return rootCause;
+ }
+
+ public String getSchemaName() {
+ return schemaName;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public String getFamilyName() {
+ return familyName;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public SQLExceptionCode getCode() {
+ return code;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/exception/UnknownFunctionException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/UnknownFunctionException.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/UnknownFunctionException.java
new file mode 100644
index 0000000..1f66992
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/UnknownFunctionException.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.exception;
+
+/**
+ * Thrown by ParseNodeFactory when it could not identify a node as a valid function.
+ */
+public class UnknownFunctionException extends RuntimeException {
+ private static final long serialVersionUID = 1L;
+ private final String funcName;
+
+ public UnknownFunctionException(String funcName) {
+ super();
+ this.funcName = funcName;
+ }
+
+ public String getFuncName() {
+ return funcName;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/exception/ValueTypeIncompatibleException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/exception/ValueTypeIncompatibleException.java b/phoenix-core/src/main/java/org/apache/phoenix/exception/ValueTypeIncompatibleException.java
new file mode 100644
index 0000000..403c009
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/exception/ValueTypeIncompatibleException.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.exception;
+
+import org.apache.phoenix.schema.IllegalDataException;
+import org.apache.phoenix.schema.PDataType;
+
+
+public class ValueTypeIncompatibleException extends IllegalDataException {
+ private static final long serialVersionUID = 1L;
+ private static SQLExceptionCode code = SQLExceptionCode.DATA_INCOMPATIBLE_WITH_TYPE;
+
+ public ValueTypeIncompatibleException(PDataType type, Integer precision, Integer scale) {
+ super(new SQLExceptionInfo.Builder(code).setMessage(getTypeDisplayString(type, precision, scale))
+ .build().toString());
+ }
+
+ private static String getTypeDisplayString(PDataType type, Integer precision, Integer scale) {
+ return type.toString() + "(" + precision + "," + scale + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java
new file mode 100644
index 0000000..b16a4e1
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java
@@ -0,0 +1,171 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.OrderByExpression;
+import org.apache.phoenix.expression.RowKeyExpression;
+import org.apache.phoenix.expression.aggregator.Aggregators;
+import org.apache.phoenix.iterate.AggregatingResultIterator;
+import org.apache.phoenix.iterate.ConcatResultIterator;
+import org.apache.phoenix.iterate.DistinctAggregatingResultIterator;
+import org.apache.phoenix.iterate.FilterAggregatingResultIterator;
+import org.apache.phoenix.iterate.GroupedAggregatingResultIterator;
+import org.apache.phoenix.iterate.LimitingResultIterator;
+import org.apache.phoenix.iterate.MergeSortRowKeyResultIterator;
+import org.apache.phoenix.iterate.OrderedAggregatingResultIterator;
+import org.apache.phoenix.iterate.OrderedResultIterator;
+import org.apache.phoenix.iterate.ParallelIterators;
+import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
+import org.apache.phoenix.iterate.PeekingResultIterator;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.iterate.SequenceResultIterator;
+import org.apache.phoenix.iterate.SpoolingResultIterator;
+import org.apache.phoenix.iterate.UngroupedAggregatingResultIterator;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.TableRef;
+
+
+
+/**
+ *
+ * Query plan for aggregating queries
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public class AggregatePlan extends BasicQueryPlan {
+ private final Aggregators aggregators;
+ private final Expression having;
+ private List<KeyRange> splits;
+
+ public AggregatePlan(
+ StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector,
+ Integer limit, OrderBy orderBy, ParallelIteratorFactory parallelIteratorFactory, GroupBy groupBy,
+ Expression having) {
+ super(context, statement, table, projector, context.getBindManager().getParameterMetaData(), limit, orderBy, groupBy, parallelIteratorFactory);
+ this.having = having;
+ this.aggregators = context.getAggregationManager().getAggregators();
+ }
+
+ @Override
+ public List<KeyRange> getSplits() {
+ return splits;
+ }
+
+ private static class OrderingResultIteratorFactory implements ParallelIteratorFactory {
+ private final QueryServices services;
+
+ public OrderingResultIteratorFactory(QueryServices services) {
+ this.services = services;
+ }
+ @Override
+ public PeekingResultIterator newIterator(ResultIterator scanner) throws SQLException {
+ Expression expression = RowKeyExpression.INSTANCE;
+ OrderByExpression orderByExpression = new OrderByExpression(expression, false, true);
+ int threshold = services.getProps().getInt(QueryServices.SPOOL_THRESHOLD_BYTES_ATTRIB, QueryServicesOptions.DEFAULT_SPOOL_THRESHOLD_BYTES);
+ return new OrderedResultIterator(scanner, Collections.<OrderByExpression>singletonList(orderByExpression), threshold);
+ }
+ }
+
+ private static class WrappingResultIteratorFactory implements ParallelIteratorFactory {
+ private final ParallelIteratorFactory innerFactory;
+ private final ParallelIteratorFactory outerFactory;
+
+ public WrappingResultIteratorFactory(ParallelIteratorFactory innerFactory, ParallelIteratorFactory outerFactory) {
+ this.innerFactory = innerFactory;
+ this.outerFactory = outerFactory;
+ }
+ @Override
+ public PeekingResultIterator newIterator(ResultIterator scanner) throws SQLException {
+ PeekingResultIterator iterator = innerFactory.newIterator(scanner);
+ return outerFactory.newIterator(iterator);
+ }
+ }
+
+ private ParallelIteratorFactory wrapParallelIteratorFactory () {
+ ParallelIteratorFactory innerFactory;
+ QueryServices services = context.getConnection().getQueryServices();
+ if (groupBy.isEmpty() || groupBy.isOrderPreserving()) {
+ innerFactory = new SpoolingResultIterator.SpoolingResultIteratorFactory(services);
+ } else {
+ innerFactory = new OrderingResultIteratorFactory(services);
+ }
+ if (parallelIteratorFactory == null) {
+ return innerFactory;
+ }
+ // wrap any existing parallelIteratorFactory
+ return new WrappingResultIteratorFactory(innerFactory, parallelIteratorFactory);
+ }
+
+ @Override
+ protected ResultIterator newIterator() throws SQLException {
+ if (groupBy.isEmpty()) {
+ UngroupedAggregateRegionObserver.serializeIntoScan(context.getScan());
+ }
+ ParallelIterators parallelIterators = new ParallelIterators(context, tableRef, statement, projection, groupBy, null, wrapParallelIteratorFactory());
+ splits = parallelIterators.getSplits();
+
+ AggregatingResultIterator aggResultIterator;
+ // No need to merge sort for ungrouped aggregation
+ if (groupBy.isEmpty()) {
+ aggResultIterator = new UngroupedAggregatingResultIterator(new ConcatResultIterator(parallelIterators), aggregators);
+ } else {
+ aggResultIterator = new GroupedAggregatingResultIterator(new MergeSortRowKeyResultIterator(parallelIterators), aggregators);
+ }
+
+ if (having != null) {
+ aggResultIterator = new FilterAggregatingResultIterator(aggResultIterator, having);
+ }
+
+ if (statement.isDistinct() && statement.isAggregate()) { // Dedup on client if select distinct and aggregation
+ aggResultIterator = new DistinctAggregatingResultIterator(aggResultIterator, getProjector());
+ }
+
+ ResultIterator resultScanner = aggResultIterator;
+ if (orderBy.getOrderByExpressions().isEmpty()) {
+ if (limit != null) {
+ resultScanner = new LimitingResultIterator(aggResultIterator, limit);
+ }
+ } else {
+ int thresholdBytes = getConnectionQueryServices(context.getConnection().getQueryServices()).getProps().getInt(
+ QueryServices.SPOOL_THRESHOLD_BYTES_ATTRIB, QueryServicesOptions.DEFAULT_SPOOL_THRESHOLD_BYTES);
+ resultScanner = new OrderedAggregatingResultIterator(aggResultIterator, orderBy.getOrderByExpressions(), thresholdBytes, limit);
+ }
+
+ if (context.getSequenceManager().getSequenceCount() > 0) {
+ resultScanner = new SequenceResultIterator(resultScanner, context.getSequenceManager());
+ }
+ return resultScanner;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/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
new file mode 100644
index 0000000..2656aaf
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/BasicQueryPlan.java
@@ -0,0 +1,197 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+import java.sql.ParameterMetaData;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+
+import com.google.common.collect.Lists;
+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.QueryPlan;
+import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.ScanRanges;
+import org.apache.phoenix.compile.StatementContext;
+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.query.ConnectionQueryServices;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.SQLCloseable;
+import org.apache.phoenix.util.SQLCloseables;
+import org.apache.phoenix.util.ScanUtil;
+
+
+
+/**
+ *
+ * Query plan that has no child plans
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public abstract class BasicQueryPlan implements QueryPlan {
+ protected static final long DEFAULT_ESTIMATED_SIZE = 10 * 1024; // 10 K
+
+ protected final TableRef tableRef;
+ protected final StatementContext context;
+ protected final FilterableStatement statement;
+ protected final RowProjector projection;
+ protected final ParameterMetaData paramMetaData;
+ protected final Integer limit;
+ protected final OrderBy orderBy;
+ protected final GroupBy groupBy;
+ protected final ParallelIteratorFactory parallelIteratorFactory;
+
+ protected BasicQueryPlan(
+ StatementContext context, FilterableStatement statement, TableRef table,
+ RowProjector projection, ParameterMetaData paramMetaData, Integer limit, OrderBy orderBy,
+ GroupBy groupBy, ParallelIteratorFactory parallelIteratorFactory) {
+ this.context = context;
+ this.statement = statement;
+ this.tableRef = table;
+ this.projection = projection;
+ this.paramMetaData = paramMetaData;
+ this.limit = limit;
+ this.orderBy = orderBy;
+ this.groupBy = groupBy;
+ this.parallelIteratorFactory = parallelIteratorFactory;
+ }
+
+ @Override
+ public GroupBy getGroupBy() {
+ return groupBy;
+ }
+
+
+ @Override
+ public OrderBy getOrderBy() {
+ return orderBy;
+ }
+
+ @Override
+ public TableRef getTableRef() {
+ return tableRef;
+ }
+
+ @Override
+ public Integer getLimit() {
+ return limit;
+ }
+
+ @Override
+ public RowProjector getProjector() {
+ return projection;
+ }
+
+ protected ConnectionQueryServices getConnectionQueryServices(ConnectionQueryServices services) {
+ // Get child services associated with tenantId of query.
+ ConnectionQueryServices childServices = context.getConnection().getTenantId() == null ?
+ services :
+ services.getChildQueryServices(new ImmutableBytesWritable(context.getConnection().getTenantId().getBytes()));
+ return childServices;
+ }
+
+// /**
+// * Sets up an id used to do round robin queue processing on the server
+// * @param scan
+// */
+// private void setProducer(Scan scan) {
+// byte[] producer = Bytes.toBytes(UUID.randomUUID().toString());
+// scan.setAttribute(HBaseServer.CALL_QUEUE_PRODUCER_ATTRIB_NAME, producer);
+// }
+
+ @Override
+ public final ResultIterator iterator() throws SQLException {
+ return iterator(Collections.<SQLCloseable>emptyList());
+ }
+
+ public final ResultIterator iterator(final List<SQLCloseable> dependencies) throws SQLException {
+ if (context.getScanRanges() == ScanRanges.NOTHING) {
+ return ResultIterator.EMPTY_ITERATOR;
+ }
+
+ Scan scan = context.getScan();
+ // Set producer on scan so HBase server does round robin processing
+ //setProducer(scan);
+ // Set the time range on the scan so we don't get back rows newer than when the statement was compiled
+ // The time stamp comes from the server at compile time when the meta data
+ // is resolved.
+ // TODO: include time range in explain plan?
+ PhoenixConnection connection = context.getConnection();
+ Long scn = connection.getSCN();
+ ScanUtil.setTimeRange(scan, scn == null ? context.getCurrentTime() : scn);
+ ScanUtil.setTenantId(scan, connection.getTenantId() == null ? null : connection.getTenantId().getBytes());
+ ResultIterator iterator = newIterator();
+ return dependencies.isEmpty() ?
+ iterator : new DelegateResultIterator(iterator) {
+ @Override
+ public void close() throws SQLException {
+ try {
+ super.close();
+ } finally {
+ SQLCloseables.closeAll(dependencies);
+ }
+ }
+ };
+ }
+
+ abstract protected ResultIterator newIterator() throws SQLException;
+
+ @Override
+ public long getEstimatedSize() {
+ return DEFAULT_ESTIMATED_SIZE;
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() {
+ return paramMetaData;
+ }
+
+ @Override
+ public FilterableStatement getStatement() {
+ return statement;
+ }
+
+ @Override
+ public StatementContext getContext() {
+ return context;
+ }
+
+ @Override
+ public ExplainPlan getExplainPlan() throws SQLException {
+ if (context.getScanRanges() == ScanRanges.NOTHING) {
+ return new ExplainPlan(Collections.singletonList("DEGENERATE SCAN OVER " + tableRef.getTable().getName().getString()));
+ }
+
+ ResultIterator iterator = iterator();
+ List<String> planSteps = Lists.newArrayListWithExpectedSize(5);
+ iterator.explain(planSteps);
+ return new ExplainPlan(planSteps);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/execute/CommitException.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/CommitException.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/CommitException.java
new file mode 100644
index 0000000..4ca47ec
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/CommitException.java
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+import java.sql.SQLException;
+
+public class CommitException extends SQLException {
+ private static final long serialVersionUID = 1L;
+ private final MutationState uncommittedState;
+ private final MutationState committedState;
+
+ public CommitException(Exception e, MutationState uncommittedState, MutationState committedState) {
+ super(e);
+ this.uncommittedState = uncommittedState;
+ this.committedState = committedState;
+ }
+
+ public MutationState getUncommittedState() {
+ return uncommittedState;
+ }
+
+ public MutationState getCommittedState() {
+ return committedState;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/execute/DegenerateQueryPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/DegenerateQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/DegenerateQueryPlan.java
new file mode 100644
index 0000000..ba38fae
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/DegenerateQueryPlan.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.compile.*;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.jdbc.PhoenixParameterMetaData;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.query.*;
+import org.apache.phoenix.schema.TableRef;
+
+public class DegenerateQueryPlan extends BasicQueryPlan {
+
+ public DegenerateQueryPlan(StatementContext context, FilterableStatement statement, TableRef table) {
+ super(context, statement, table, RowProjector.EMPTY_PROJECTOR, PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA, null, OrderBy.EMPTY_ORDER_BY, GroupBy.EMPTY_GROUP_BY, null);
+ context.setScanRanges(ScanRanges.NOTHING);
+ }
+
+ @Override
+ public List<KeyRange> getSplits() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ protected ResultIterator newIterator() throws SQLException {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
new file mode 100644
index 0000000..73b456a
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
@@ -0,0 +1,186 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+import java.sql.ParameterMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import org.apache.hadoop.hbase.client.Scan;
+
+import com.google.common.collect.Lists;
+import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
+import org.apache.phoenix.compile.ExplainPlan;
+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.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.job.JobManager.JobCallable;
+import org.apache.phoenix.join.HashCacheClient;
+import org.apache.phoenix.join.HashJoinInfo;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.SQLCloseable;
+
+public class HashJoinPlan implements QueryPlan {
+
+ private BasicQueryPlan plan;
+ private HashJoinInfo joinInfo;
+ private List<Expression>[] hashExpressions;
+ private QueryPlan[] hashPlans;
+
+ public HashJoinPlan(BasicQueryPlan plan, HashJoinInfo joinInfo,
+ List<Expression>[] hashExpressions, QueryPlan[] hashPlans) {
+ this.plan = plan;
+ this.joinInfo = joinInfo;
+ this.hashExpressions = hashExpressions;
+ this.hashPlans = hashPlans;
+ }
+
+ @Override
+ public Integer getLimit() {
+ return plan.getLimit();
+ }
+
+ @Override
+ public OrderBy getOrderBy() {
+ return plan.getOrderBy();
+ }
+
+ @Override
+ public RowProjector getProjector() {
+ return plan.getProjector();
+ }
+
+ @Override
+ public ResultIterator iterator() throws SQLException {
+ ImmutableBytesPtr[] joinIds = joinInfo.getJoinIds();
+ assert (joinIds.length == hashExpressions.length && joinIds.length == hashPlans.length);
+
+ final HashCacheClient hashClient = new HashCacheClient(plan.getContext().getConnection());
+ Scan scan = plan.getContext().getScan();
+ final ScanRanges ranges = plan.getContext().getScanRanges();
+
+ int count = joinIds.length;
+ ConnectionQueryServices services = getContext().getConnection().getQueryServices();
+ ExecutorService executor = services.getExecutor();
+ List<Future<ServerCache>> futures = new ArrayList<Future<ServerCache>>(count);
+ List<SQLCloseable> dependencies = new ArrayList<SQLCloseable>(count);
+ for (int i = 0; i < count; i++) {
+ final int index = i;
+ futures.add(executor.submit(new JobCallable<ServerCache>() {
+
+ @Override
+ public ServerCache call() throws Exception {
+ QueryPlan hashPlan = hashPlans[index];
+ return hashClient.addHashCache(ranges, hashPlan.iterator(),
+ hashPlan.getEstimatedSize(), hashExpressions[index], plan.getTableRef());
+ }
+
+ @Override
+ public Object getJobId() {
+ return HashJoinPlan.this;
+ }
+ }));
+ }
+ for (int i = 0; i < count; i++) {
+ try {
+ ServerCache cache = futures.get(i).get();
+ joinIds[i].set(cache.getId());
+ dependencies.add(cache);
+ } catch (InterruptedException e) {
+ throw new SQLException("Hash join execution interrupted.", e);
+ } catch (ExecutionException e) {
+ throw new SQLException("Encountered exception in hash plan execution.",
+ e.getCause());
+ }
+ }
+ HashJoinInfo.serializeHashJoinIntoScan(scan, joinInfo);
+
+ return plan.iterator(dependencies);
+ }
+
+ @Override
+ public long getEstimatedSize() {
+ return plan.getEstimatedSize();
+ }
+
+ @Override
+ public List<KeyRange> getSplits() {
+ return plan.getSplits();
+ }
+
+ @Override
+ public ExplainPlan getExplainPlan() throws SQLException {
+ List<String> mainQuerySteps = plan.getExplainPlan().getPlanSteps();
+ List<String> planSteps = Lists.newArrayList(mainQuerySteps);
+ int count = hashPlans.length;
+ planSteps.add(" PARALLEL EQUI-JOIN " + count + " HASH TABLES:");
+ for (int i = 0; i < count; i++) {
+ boolean skipMerge = joinInfo.getSchemas()[i].getFieldCount() == 0;
+ planSteps.add(" BUILD HASH TABLE " + i + (skipMerge ? " (SKIP MERGE)" : ""));
+ List<String> steps = hashPlans[i].getExplainPlan().getPlanSteps();
+ for (String step : steps) {
+ planSteps.add(" " + step);
+ }
+ }
+ if (joinInfo.getPostJoinFilterExpression() != null) {
+ planSteps.add(" AFTER-JOIN SERVER FILTER BY " + joinInfo.getPostJoinFilterExpression().toString());
+ }
+
+ return new ExplainPlan(planSteps);
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() {
+ return plan.getParameterMetaData();
+ }
+
+ @Override
+ public StatementContext getContext() {
+ return plan.getContext();
+ }
+
+ @Override
+ public GroupBy getGroupBy() {
+ return plan.getGroupBy();
+ }
+
+ @Override
+ public TableRef getTableRef() {
+ return plan.getTableRef();
+ }
+
+ @Override
+ public FilterableStatement getStatement() {
+ return plan.getStatement();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/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
new file mode 100644
index 0000000..298c7c1
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -0,0 +1,444 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.hadoop.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.cache.ServerCacheClient;
+import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
+import org.apache.phoenix.coprocessor.MetaDataProtocol.MetaDataMutationResult;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.index.IndexMaintainer;
+import org.apache.phoenix.index.IndexMetaDataCacheClient;
+import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.IllegalDataException;
+import org.apache.phoenix.schema.MetaDataClient;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PRow;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.ByteUtil;
+import org.apache.phoenix.util.IndexUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.SQLCloseable;
+import org.apache.phoenix.util.ServerUtil;
+
+/**
+ *
+ * Tracks the uncommitted state
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public class MutationState implements SQLCloseable {
+ private static final Logger logger = LoggerFactory.getLogger(MutationState.class);
+
+ private PhoenixConnection connection;
+ private final long maxSize;
+ private final ImmutableBytesPtr tempPtr = new ImmutableBytesPtr();
+ private final Map<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> mutations = Maps.newHashMapWithExpectedSize(3); // TODO: Sizing?
+ private final long sizeOffset;
+ private int numRows = 0;
+
+ public MutationState(int maxSize, PhoenixConnection connection) {
+ this(maxSize,connection,0);
+ }
+
+ public MutationState(int maxSize, PhoenixConnection connection, long sizeOffset) {
+ this.maxSize = maxSize;
+ this.connection = connection;
+ this.sizeOffset = sizeOffset;
+ }
+
+ public MutationState(TableRef table, Map<ImmutableBytesPtr,Map<PColumn,byte[]>> mutations, long sizeOffset, long maxSize, PhoenixConnection connection) {
+ this.maxSize = maxSize;
+ this.connection = connection;
+ this.mutations.put(table, mutations);
+ this.sizeOffset = sizeOffset;
+ this.numRows = mutations.size();
+ throwIfTooBig();
+ }
+
+ private MutationState(List<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> entries, long sizeOffset, long maxSize, PhoenixConnection connection) {
+ this.maxSize = maxSize;
+ this.connection = connection;
+ this.sizeOffset = sizeOffset;
+ for (Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> entry : entries) {
+ numRows += entry.getValue().size();
+ this.mutations.put(entry.getKey(), entry.getValue());
+ }
+ throwIfTooBig();
+ }
+
+ private void throwIfTooBig() {
+ if (numRows > maxSize) {
+ // TODO: throw SQLException ?
+ throw new IllegalArgumentException("MutationState size of " + numRows + " is bigger than max allowed size of " + maxSize);
+ }
+ }
+
+ public long getUpdateCount() {
+ return sizeOffset + numRows;
+ }
+ /**
+ * Combine a newer mutation with this one, where in the event of overlaps,
+ * the newer one will take precedence.
+ * @param newMutation the newer mutation
+ */
+ public void join(MutationState newMutation) {
+ if (this == newMutation) { // Doesn't make sense
+ return;
+ }
+ // Merge newMutation with this one, keeping state from newMutation for any overlaps
+ for (Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> entry : newMutation.mutations.entrySet()) {
+ // Replace existing entries for the table with new entries
+ Map<ImmutableBytesPtr,Map<PColumn,byte[]>> existingRows = this.mutations.put(entry.getKey(), entry.getValue());
+ if (existingRows != null) { // Rows for that table already exist
+ // Loop through new rows and replace existing with new
+ for (Map.Entry<ImmutableBytesPtr,Map<PColumn,byte[]>> rowEntry : entry.getValue().entrySet()) {
+ // Replace existing row with new row
+ Map<PColumn,byte[]> existingValues = existingRows.put(rowEntry.getKey(), rowEntry.getValue());
+ if (existingValues != null) {
+ if (existingValues != PRow.DELETE_MARKER) {
+ Map<PColumn,byte[]> newRow = rowEntry.getValue();
+ // if new row is PRow.DELETE_MARKER, it means delete, and we don't need to merge it with existing row.
+ if (newRow != PRow.DELETE_MARKER) {
+ // Replace existing column values with new column values
+ for (Map.Entry<PColumn,byte[]> valueEntry : newRow.entrySet()) {
+ existingValues.put(valueEntry.getKey(), valueEntry.getValue());
+ }
+ // Now that the existing row has been merged with the new row, replace it back
+ // again (since it was replaced with the new one above).
+ existingRows.put(rowEntry.getKey(), existingValues);
+ }
+ }
+ } else {
+ numRows++;
+ }
+ }
+ // Put the existing one back now that it's merged
+ this.mutations.put(entry.getKey(), existingRows);
+ } else {
+ numRows += entry.getValue().size();
+ }
+ }
+ throwIfTooBig();
+ }
+
+ private Iterator<Pair<byte[],List<Mutation>>> addRowMutations(final TableRef tableRef, final Map<ImmutableBytesPtr, Map<PColumn, byte[]>> values, long timestamp, boolean includeMutableIndexes) {
+ final List<Mutation> mutations = Lists.newArrayListWithExpectedSize(values.size());
+ Iterator<Map.Entry<ImmutableBytesPtr,Map<PColumn,byte[]>>> iterator = values.entrySet().iterator();
+ while (iterator.hasNext()) {
+ Map.Entry<ImmutableBytesPtr,Map<PColumn,byte[]>> rowEntry = iterator.next();
+ ImmutableBytesPtr key = rowEntry.getKey();
+ PRow row = tableRef.getTable().newRow(connection.getKeyValueBuilder(), timestamp, key);
+ if (rowEntry.getValue() == PRow.DELETE_MARKER) { // means delete
+ row.delete();
+ } else {
+ for (Map.Entry<PColumn,byte[]> valueEntry : rowEntry.getValue().entrySet()) {
+ row.setValue(valueEntry.getKey(), valueEntry.getValue());
+ }
+ }
+ mutations.addAll(row.toRowMutations());
+ }
+ final Iterator<PTable> indexes = // Only maintain tables with immutable rows through this client-side mechanism
+ (tableRef.getTable().isImmutableRows() || includeMutableIndexes) ?
+ IndexMaintainer.nonDisabledIndexIterator(tableRef.getTable().getIndexes().iterator()) :
+ Iterators.<PTable>emptyIterator();
+ return new Iterator<Pair<byte[],List<Mutation>>>() {
+ boolean isFirst = true;
+
+ @Override
+ public boolean hasNext() {
+ return isFirst || indexes.hasNext();
+ }
+
+ @Override
+ public Pair<byte[], List<Mutation>> next() {
+ if (isFirst) {
+ isFirst = false;
+ return new Pair<byte[],List<Mutation>>(tableRef.getTable().getPhysicalName().getBytes(),mutations);
+ }
+ PTable index = indexes.next();
+ List<Mutation> indexMutations;
+ try {
+ indexMutations =
+ IndexUtil.generateIndexData(tableRef.getTable(), index, mutations,
+ tempPtr, connection.getKeyValueBuilder());
+ } catch (SQLException e) {
+ throw new IllegalDataException(e);
+ }
+ return new Pair<byte[],List<Mutation>>(index.getPhysicalName().getBytes(),indexMutations);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ };
+ }
+
+ /**
+ * Get the unsorted list of HBase mutations for the tables with uncommitted data.
+ * @return list of HBase mutations for uncommitted data.
+ */
+ public Iterator<Pair<byte[],List<Mutation>>> toMutations() {
+ return toMutations(false);
+ }
+
+ public Iterator<Pair<byte[],List<Mutation>>> toMutations(final boolean includeMutableIndexes) {
+ final Iterator<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> iterator = this.mutations.entrySet().iterator();
+ if (!iterator.hasNext()) {
+ return Iterators.emptyIterator();
+ }
+ Long scn = connection.getSCN();
+ final long timestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
+ return new Iterator<Pair<byte[],List<Mutation>>>() {
+ private Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> current = iterator.next();
+ private Iterator<Pair<byte[],List<Mutation>>> innerIterator = init();
+
+ private Iterator<Pair<byte[],List<Mutation>>> init() {
+ return addRowMutations(current.getKey(), current.getValue(), timestamp, includeMutableIndexes);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return innerIterator.hasNext() || iterator.hasNext();
+ }
+
+ @Override
+ public Pair<byte[], List<Mutation>> next() {
+ if (!innerIterator.hasNext()) {
+ current = iterator.next();
+ }
+ return innerIterator.next();
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+
+ };
+ }
+
+ /**
+ * Validates that the meta data is still valid based on the current server time
+ * and returns the server time to use for the upsert for each table.
+ * @param connection
+ * @return the server time to use for the upsert
+ * @throws SQLException if the table or any columns no longer exist
+ */
+ private long[] validate() throws SQLException {
+ int i = 0;
+ Long scn = connection.getSCN();
+ MetaDataClient client = new MetaDataClient(connection);
+ long[] timeStamps = new long[this.mutations.size()];
+ for (Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> entry : mutations.entrySet()) {
+ TableRef tableRef = entry.getKey();
+ long serverTimeStamp = tableRef.getTimeStamp();
+ PTable table = tableRef.getTable();
+ if (!connection.getAutoCommit()) {
+ MetaDataMutationResult result = client.updateCache(table.getSchemaName().getString(), table.getTableName().getString());
+ long timestamp = result.getMutationTime();
+ if (timestamp != QueryConstants.UNSET_TIMESTAMP) {
+ serverTimeStamp = timestamp;
+ if (result.wasUpdated()) {
+ // TODO: use bitset?
+ PColumn[] columns = new PColumn[table.getColumns().size()];
+ for (Map.Entry<ImmutableBytesPtr,Map<PColumn,byte[]>> rowEntry : entry.getValue().entrySet()) {
+ Map<PColumn,byte[]> valueEntry = rowEntry.getValue();
+ if (valueEntry != PRow.DELETE_MARKER) {
+ for (PColumn column : valueEntry.keySet()) {
+ columns[column.getPosition()] = column;
+ }
+ }
+ }
+ table = connection.getPMetaData().getTable(tableRef.getTable().getName().getString());
+ for (PColumn column : columns) {
+ if (column != null) {
+ table.getColumnFamily(column.getFamilyName().getString()).getColumn(column.getName().getString());
+ }
+ }
+ }
+ }
+ }
+ timeStamps[i++] = scn == null ? serverTimeStamp : scn;
+ }
+ return timeStamps;
+ }
+
+ private static void logMutationSize(HTableInterface htable, List<Mutation> mutations) {
+ long byteSize = 0;
+ int keyValueCount = 0;
+ for (Mutation mutation : mutations) {
+ if (mutation.getFamilyMap() != null) { // Not a Delete of the row
+ for (Entry<byte[], List<KeyValue>> entry : mutation.getFamilyMap().entrySet()) {
+ if (entry.getValue() != null) {
+ for (KeyValue kv : entry.getValue()) {
+ byteSize += kv.getBuffer().length;
+ keyValueCount++;
+ }
+ }
+ }
+ }
+ }
+ logger.debug("Sending " + mutations.size() + " mutations for " + Bytes.toString(htable.getTableName()) + " with " + keyValueCount + " key values of total size " + byteSize + " bytes");
+ }
+
+ public void commit() throws SQLException {
+ int i = 0;
+ byte[] tenantId = connection.getTenantId() == null ? null : connection.getTenantId().getBytes();
+ long[] serverTimeStamps = validate();
+ Iterator<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> iterator = this.mutations.entrySet().iterator();
+ List<Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>>> committedList = Lists.newArrayListWithCapacity(this.mutations.size());
+ while (iterator.hasNext()) {
+ Map.Entry<TableRef, Map<ImmutableBytesPtr,Map<PColumn,byte[]>>> entry = iterator.next();
+ Map<ImmutableBytesPtr,Map<PColumn,byte[]>> valuesMap = entry.getValue();
+ TableRef tableRef = entry.getKey();
+ PTable table = tableRef.getTable();
+ table.getIndexMaintainers(tempPtr);
+ boolean hasIndexMaintainers = tempPtr.getLength() > 0;
+ boolean isDataTable = true;
+ long serverTimestamp = serverTimeStamps[i++];
+ Iterator<Pair<byte[],List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, serverTimestamp, false);
+ while (mutationsIterator.hasNext()) {
+ Pair<byte[],List<Mutation>> pair = mutationsIterator.next();
+ byte[] htableName = pair.getFirst();
+ List<Mutation> mutations = pair.getSecond();
+
+ int retryCount = 0;
+ boolean shouldRetry = false;
+ do {
+ ServerCache cache = null;
+ if (hasIndexMaintainers && isDataTable) {
+ byte[] attribValue = null;
+ byte[] uuidValue;
+ if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, tempPtr.getLength())) {
+ IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
+ cache = client.addIndexMetadataCache(mutations, tempPtr);
+ uuidValue = cache.getId();
+ // If we haven't retried yet, retry for this case only, as it's possible that
+ // a split will occur after we send the index metadata cache to all known
+ // region servers.
+ shouldRetry = true;
+ } else {
+ attribValue = ByteUtil.copyKeyBytesIfNecessary(tempPtr);
+ uuidValue = ServerCacheClient.generateId();
+ }
+ // Either set the UUID to be able to access the index metadata from the cache
+ // or set the index metadata directly on the Mutation
+ for (Mutation mutation : mutations) {
+ if (tenantId != null) {
+ mutation.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
+ }
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ if (attribValue != null) {
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
+ }
+ }
+ }
+
+ SQLException sqlE = null;
+ HTableInterface hTable = connection.getQueryServices().getTable(htableName);
+ try {
+ if (logger.isDebugEnabled()) logMutationSize(hTable, mutations);
+ long startTime = System.currentTimeMillis();
+ hTable.batch(mutations);
+ shouldRetry = false;
+ if (logger.isDebugEnabled()) logger.debug("Total time for batch call of " + mutations.size() + " mutations into " + table.getName().getString() + ": " + (System.currentTimeMillis() - startTime) + " ms");
+ committedList.add(entry);
+ } catch (Exception e) {
+ SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
+ if (inferredE != null) {
+ if (shouldRetry && retryCount == 0 && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND.getErrorCode()) {
+ // Swallow this exception once, as it's possible that we split after sending the index metadata
+ // and one of the region servers doesn't have it. This will cause it to have it the next go around.
+ // If it fails again, we don't retry.
+ logger.warn("Swallowing exception and retrying after clearing meta cache on connection. " + inferredE);
+ connection.getQueryServices().clearTableRegionCache(htableName);
+ continue;
+ }
+ e = inferredE;
+ }
+ // Throw to client with both what was committed so far and what is left to be committed.
+ // That way, client can either undo what was done or try again with what was not done.
+ sqlE = new CommitException(e, this, new MutationState(committedList, this.sizeOffset, this.maxSize, this.connection));
+ } finally {
+ try {
+ hTable.close();
+ } catch (IOException e) {
+ if (sqlE != null) {
+ sqlE.setNextException(ServerUtil.parseServerException(e));
+ } else {
+ sqlE = ServerUtil.parseServerException(e);
+ }
+ } finally {
+ try {
+ if (cache != null) {
+ cache.close();
+ }
+ } finally {
+ if (sqlE != null) {
+ throw sqlE;
+ }
+ }
+ }
+ }
+ } while (shouldRetry && retryCount++ < 1);
+ isDataTable = false;
+ }
+ numRows -= entry.getValue().size();
+ iterator.remove(); // Remove batches as we process them
+ }
+ assert(numRows==0);
+ assert(this.mutations.isEmpty());
+ }
+
+ public void rollback(PhoenixConnection connection) throws SQLException {
+ this.mutations.clear();
+ numRows = 0;
+ }
+
+ @Override
+ public void close() throws SQLException {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/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
new file mode 100644
index 0000000..01199c9
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/ScanPlan.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.execute;
+
+
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.coprocessor.ScanRegionObserver;
+import org.apache.phoenix.iterate.ConcatResultIterator;
+import org.apache.phoenix.iterate.LimitingResultIterator;
+import org.apache.phoenix.iterate.MergeSortRowKeyResultIterator;
+import org.apache.phoenix.iterate.MergeSortTopNResultIterator;
+import org.apache.phoenix.iterate.ParallelIterators;
+import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.iterate.SequenceResultIterator;
+import org.apache.phoenix.iterate.SpoolingResultIterator;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.query.QueryServices;
+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.util.ScanUtil;
+
+
+
+/**
+ *
+ * Query plan for a basic table scan
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public class ScanPlan extends BasicQueryPlan {
+ private List<KeyRange> splits;
+
+ public ScanPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector, Integer limit, OrderBy orderBy, ParallelIteratorFactory parallelIteratorFactory) {
+ super(context, statement, table, projector, context.getBindManager().getParameterMetaData(), limit, orderBy, null, parallelIteratorFactory == null ? new SpoolingResultIterator.SpoolingResultIteratorFactory(context.getConnection().getQueryServices()) : parallelIteratorFactory);
+ if (!orderBy.getOrderByExpressions().isEmpty()) { // TopN
+ int thresholdBytes = context.getConnection().getQueryServices().getProps().getInt(
+ QueryServices.SPOOL_THRESHOLD_BYTES_ATTRIB, QueryServicesOptions.DEFAULT_SPOOL_THRESHOLD_BYTES);
+ ScanRegionObserver.serializeIntoScan(context.getScan(), thresholdBytes, limit == null ? -1 : limit, orderBy.getOrderByExpressions(), projector.getEstimatedRowByteSize());
+ }
+ }
+
+ @Override
+ public List<KeyRange> getSplits() {
+ return splits;
+ }
+
+ @Override
+ protected ResultIterator newIterator() throws SQLException {
+ // Set any scan attributes before creating the scanner, as it will be too late afterwards
+ context.getScan().setAttribute(ScanRegionObserver.NON_AGGREGATE_QUERY, QueryConstants.TRUE);
+ if (OrderBy.REV_ROW_KEY_ORDER_BY.equals(orderBy)) {
+ ScanUtil.setReversed(context.getScan());
+ }
+ ResultIterator scanner;
+ TableRef tableRef = this.getTableRef();
+ PTable table = tableRef.getTable();
+ boolean isSalted = table.getBucketNum() != null;
+ /* If no limit or topN, use parallel iterator so that we get results faster. Otherwise, if
+ * limit is provided, run query serially.
+ */
+ boolean isOrdered = !orderBy.getOrderByExpressions().isEmpty();
+ ParallelIterators iterators = new ParallelIterators(context, tableRef, statement, projection, GroupBy.EMPTY_GROUP_BY, isOrdered ? null : limit, parallelIteratorFactory);
+ splits = iterators.getSplits();
+ if (isOrdered) {
+ scanner = new MergeSortTopNResultIterator(iterators, limit, orderBy.getOrderByExpressions());
+ } else {
+ if (isSalted &&
+ (getConnectionQueryServices(context.getConnection().getQueryServices()).getProps().getBoolean(
+ QueryServices.ROW_KEY_ORDER_SALTED_TABLE_ATTRIB,
+ QueryServicesOptions.DEFAULT_ROW_KEY_ORDER_SALTED_TABLE) ||
+ orderBy == OrderBy.FWD_ROW_KEY_ORDER_BY ||
+ orderBy == OrderBy.REV_ROW_KEY_ORDER_BY)) { // ORDER BY was optimized out b/c query is in row key order
+ scanner = new MergeSortRowKeyResultIterator(iterators, SaltingUtil.NUM_SALTING_BYTES, orderBy == OrderBy.REV_ROW_KEY_ORDER_BY);
+ } else {
+ scanner = new ConcatResultIterator(iterators);
+ }
+ if (limit != null) {
+ scanner = new LimitingResultIterator(scanner, limit);
+ }
+ }
+
+ if (context.getSequenceManager().getSequenceCount() > 0) {
+ scanner = new SequenceResultIterator(scanner, context.getSequenceManager());
+ }
+ return scanner;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/expression/AddExpression.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/AddExpression.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/AddExpression.java
new file mode 100644
index 0000000..fd906c5
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/AddExpression.java
@@ -0,0 +1,56 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.util.List;
+
+import org.apache.phoenix.expression.visitor.ExpressionVisitor;
+
+
+/**
+ *
+ * Subtract expression implementation
+ *
+ * @author jtaylor
+ * @since 0.1
+ */
+public abstract class AddExpression extends BaseAddSubtractExpression {
+ public AddExpression() {
+ }
+
+ public AddExpression(List<Expression> children) {
+ super(children);
+ }
+
+ @Override
+ public final <T> T accept(ExpressionVisitor<T> visitor) {
+ List<T> l = acceptChildren(visitor, visitor.visitEnter(this));
+ T t = visitor.visitLeave(this, l);
+ if (t == null) {
+ t = visitor.defaultReturn(this, l);
+ }
+ return t;
+ }
+
+ @Override
+ public String getOperatorString() {
+ return " + ";
+ }
+}