You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by td...@apache.org on 2015/11/10 02:18:48 UTC
[33/50] [abbrv] phoenix git commit: merge master
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/optimize/QueryOptimizer.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServices.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 13a0559,b718089..db799da
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@@ -107,12 -109,12 +114,13 @@@ import org.apache.phoenix.exception.SQL
import org.apache.phoenix.execute.MutationState;
import org.apache.phoenix.hbase.index.IndexRegionSplitPolicy;
import org.apache.phoenix.hbase.index.Indexer;
-import org.apache.phoenix.hbase.index.covered.CoveredColumnsIndexBuilder;
+import org.apache.phoenix.hbase.index.covered.NonTxIndexBuilder;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.hbase.index.util.KeyValueBuilder;
+ import org.apache.phoenix.hbase.index.util.VersionUtil;
import org.apache.phoenix.index.PhoenixIndexBuilder;
import org.apache.phoenix.index.PhoenixIndexCodec;
+import org.apache.phoenix.index.PhoenixTransactionalIndexer;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.jdbc.PhoenixEmbeddedDriver.ConnectionInfo;
@@@ -163,8 -166,7 +177,9 @@@ import org.apache.twill.zookeeper.ZKCli
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.base.Objects;
+import com.google.common.base.Stopwatch;
+ import com.google.common.base.Joiner;
import com.google.common.base.Throwables;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
@@@ -202,9 -202,9 +216,10 @@@ public class ConnectionQueryServicesImp
@GuardedBy("connectionCountLock")
private int connectionCount = 0;
private final Object connectionCountLock = new Object();
+ private final boolean returnSequenceValues ;
private HConnection connection;
+ private TransactionServiceClient txServiceClient;
private volatile boolean initialized;
private volatile int nSequenceSaltBuckets;
@@@ -279,32 -279,9 +295,33 @@@
.maximumSize(MAX_TABLE_STATS_CACHE_ENTRIES)
.expireAfterWrite(halfStatsUpdateFreq, TimeUnit.MILLISECONDS)
.build();
+ this.returnSequenceValues = config.getBoolean(QueryServices.RETURN_SEQUENCE_VALUES_ATTRIB, QueryServicesOptions.DEFAULT_RETURN_SEQUENCE_VALUES);
}
+ @Override
+ public TransactionSystemClient getTransactionSystemClient() {
+ return txServiceClient;
+ }
+
+ private void initTxServiceClient() {
+ String zkQuorumServersString = connectionInfo.getZookeeperQuorum()+":"+connectionInfo.getPort();
+ ZKClientService zkClientService = ZKClientServices.delegate(
+ ZKClients.reWatchOnExpire(
+ ZKClients.retryOnFailure(
+ ZKClientService.Builder.of(zkQuorumServersString)
+ .setSessionTimeout(config.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT))
+ .build(),
+ RetryStrategies.exponentialDelay(500, 2000, TimeUnit.MILLISECONDS)
+ )
+ )
+ );
+ zkClientService.startAndWait();
+ ZKDiscoveryService zkDiscoveryService = new ZKDiscoveryService(zkClientService);
+ PooledClientProvider pooledClientProvider = new PooledClientProvider(
+ config, zkDiscoveryService);
+ this.txServiceClient = new TransactionServiceClient(config,pooledClientProvider);
+ }
+
private void openConnection() throws SQLException {
try {
// check if we need to authenticate with kerberos
@@@ -330,9 -306,11 +347,9 @@@
@Override
public HTableInterface getTable(byte[] tableName) throws SQLException {
try {
- return HBaseFactoryProvider.getHTableFactory().getTable(tableName, connection, getExecutor());
+ return HBaseFactoryProvider.getHTableFactory().getTable(tableName, connection, null);
} catch (org.apache.hadoop.hbase.TableNotFoundException e) {
- byte[][] schemaAndTableName = new byte[2][];
- SchemaUtil.getVarChars(tableName, schemaAndTableName);
- throw new TableNotFoundException(Bytes.toString(schemaAndTableName[0]), Bytes.toString(schemaAndTableName[1]));
+ throw new TableNotFoundException(SchemaUtil.getSchemaNameFromFullName(tableName), SchemaUtil.getTableNameFromFullName(tableName));
} catch (IOException e) {
throw new SQLException(e);
}
@@@ -1787,6 -1772,15 +1860,15 @@@
return !isHColumnProperty(propName) && !TableProperty.isPhoenixTableProperty(propName);
}
+ private HashSet<String> existingColumnFamiliesForBaseTable(PName baseTableName) throws TableNotFoundException {
+ synchronized (latestMetaDataLock) {
+ throwConnectionClosedIfNullMetaData();
- PTable table = latestMetaData.getTable(new PTableKey(null, baseTableName.getString()));
++ PTable table = latestMetaData.getTableRef(new PTableKey(null, baseTableName.getString())).getTable();
+ latestMetaDataLock.notifyAll();
+ return existingColumnFamilies(table);
+ }
+ }
+
private HashSet<String> existingColumnFamilies(PTable table) {
List<PColumnFamily> cfs = table.getColumnFamilies();
HashSet<String> cfNames = new HashSet<>(cfs.size());
@@@ -1857,22 -1852,20 +1940,20 @@@
return result;
}
-
- /**
- * Keeping this to use for further upgrades. This method closes the oldMetaConnection.
+
+ /**
+ * This closes the passed connection.
*/
- private PhoenixConnection addColumnsIfNotExists(PhoenixConnection oldMetaConnection,
- String tableName, long timestamp, String columns) throws SQLException {
-
- Properties props = new Properties(oldMetaConnection.getClientInfo());
+ private PhoenixConnection addColumn(PhoenixConnection oldMetaConnection, String tableName, long timestamp, String columns, boolean addIfNotExists) throws SQLException {
+ Properties props = PropertiesUtil.deepCopy(oldMetaConnection.getClientInfo());
props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(timestamp));
// Cannot go through DriverManager or you end up in an infinite loop because it'll call init again
- PhoenixConnection metaConnection = new PhoenixConnection(this, oldMetaConnection.getURL(), props, oldMetaConnection.getMetaDataCache());
+ PhoenixConnection metaConnection = new PhoenixConnection(oldMetaConnection, this, props);
SQLException sqlE = null;
try {
- metaConnection.createStatement().executeUpdate("ALTER TABLE " + tableName + " ADD IF NOT EXISTS " + columns );
+ metaConnection.createStatement().executeUpdate("ALTER TABLE " + tableName + " ADD " + (addIfNotExists ? " IF NOT EXISTS " : "") + columns );
} catch (SQLException e) {
- logger.warn("addColumnsIfNotExists failed due to:" + e);
+ logger.warn("Add column failed due to:" + e);
sqlE = e;
} finally {
try {
@@@ -1935,32 -1936,29 +2024,32 @@@
long currentServerSideTableTimeStamp = e.getTable().getTimeStamp();
String columnsToAdd = "";
- if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_4_0) {
- if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
- // We know that we always need to add the STORE_NULLS column for 4.3 release
- columnsToAdd = ", " + PhoenixDatabaseMetaData.STORE_NULLS + " " + PBoolean.INSTANCE.getSqlTypeName();
- HBaseAdmin admin = null;
- try {
- admin = getAdmin();
- HTableDescriptor[] localIndexTables = admin.listTables(MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX+".*");
- for (HTableDescriptor table : localIndexTables) {
- if (table.getValue(MetaDataUtil.PARENT_TABLE_KEY) == null
- && table.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME) != null) {
- table.setValue(MetaDataUtil.PARENT_TABLE_KEY,
- MetaDataUtil.getUserTableName(table
- .getNameAsString()));
- // Explicitly disable, modify and enable the table to ensure co-location of data
- // and index regions. If we just modify the table descriptor when online schema
- // change enabled may reopen the region in same region server instead of following data region.
- admin.disableTable(table.getTableName());
- admin.modifyTable(table.getTableName(), table);
- admin.enableTable(table.getTableName());
++ if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_6_0) {
+ columnsToAdd = PhoenixDatabaseMetaData.TRANSACTIONAL + " " + PBoolean.INSTANCE.getSqlTypeName();
+ if(currentServerSideTableTimeStamp < MetaDataProtocol.MIN_SYSTEM_TABLE_TIMESTAMP_4_3_0) {
+ // We know that we always need to add the STORE_NULLS column for 4.3 release
+ columnsToAdd += "," + PhoenixDatabaseMetaData.STORE_NULLS + " " + PBoolean.INSTANCE.getSqlTypeName();
+ HBaseAdmin admin = null;
+ try {
+ admin = getAdmin();
+ HTableDescriptor[] localIndexTables = admin.listTables(MetaDataUtil.LOCAL_INDEX_TABLE_PREFIX+".*");
+ for (HTableDescriptor table : localIndexTables) {
+ if (table.getValue(MetaDataUtil.PARENT_TABLE_KEY) == null
+ && table.getValue(MetaDataUtil.IS_LOCAL_INDEX_TABLE_PROP_NAME) != null) {
+ table.setValue(MetaDataUtil.PARENT_TABLE_KEY,
+ MetaDataUtil.getUserTableName(table
+ .getNameAsString()));
+ // Explicitly disable, modify and enable the table to ensure co-location of data
+ // and index regions. If we just modify the table descriptor when online schema
+ // change enabled may reopen the region in same region server instead of following data region.
+ admin.disableTable(table.getTableName());
+ admin.modifyTable(table.getTableName(), table);
+ admin.enableTable(table.getTableName());
+ }
}
+ } finally {
+ if (admin != null) admin.close();
}
- } finally {
- if (admin != null) admin.close();
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index 3f31141,3fa0c1e..1102e22
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@@ -107,31 -104,13 +111,32 @@@ public class ConnectionlessQueryService
private volatile SQLException initializationException;
private final Map<String, List<HRegionLocation>> tableSplits = Maps.newHashMap();
- public ConnectionlessQueryServicesImpl(QueryServices queryServices, ConnectionInfo connInfo) {
- super(queryServices);
+ public ConnectionlessQueryServicesImpl(QueryServices services, ConnectionInfo connInfo, Properties info) {
+ super(services);
userName = connInfo.getPrincipal();
metaData = newEmptyMetaData();
+
// Use KeyValueBuilder that builds real KeyValues, as our test utils require this
this.kvBuilder = GenericKeyValueBuilder.INSTANCE;
+ // TOOD: copy/paste from ConnectionQueryServicesImpl
+ Configuration config = HBaseFactoryProvider.getConfigurationFactory().getConfiguration();
+ for (Entry<String,String> entry : services.getProps()) {
+ config.set(entry.getKey(), entry.getValue());
+ }
+ if (info != null) {
+ for (Object key : info.keySet()) {
+ config.set((String) key, info.getProperty((String) key));
+ }
+ }
+ for (Entry<String,String> entry : connInfo.asProps()) {
+ config.set(entry.getKey(), entry.getValue());
+ }
+
+ // Without making a copy of the configuration we cons up, we lose some of our properties
+ // on the server side during testing.
+ config = HBaseFactoryProvider.getConfigurationFactory().getConfiguration(config);
+ TransactionManager txnManager = new TransactionManager(config);
+ this.txSystemClient = new InMemoryTxSystemClient(txnManager);
}
private PMetaData newEmptyMetaData() {
@@@ -512,8 -492,44 +523,48 @@@
}
@Override
+ public TransactionSystemClient getTransactionSystemClient() {
+ return txSystemClient;
+ }
+
+ public MetaDataMutationResult createFunction(List<Mutation> functionData, PFunction function, boolean temporary)
+ throws SQLException {
+ return new MetaDataMutationResult(MutationCode.FUNCTION_NOT_FOUND, 0l, null);
+ }
+
+ @Override
+ public PMetaData addFunction(PFunction function) throws SQLException {
+ return metaData = this.metaData.addFunction(function);
+ }
+
+ @Override
+ public PMetaData removeFunction(PName tenantId, String function, long functionTimeStamp)
+ throws SQLException {
+ return metaData = this.metaData.removeFunction(tenantId, function, functionTimeStamp);
+ }
+
+ @Override
+ public MetaDataMutationResult getFunctions(PName tenantId,
+ List<Pair<byte[], Long>> functionNameAndTimeStampPairs, long clientTimestamp)
+ throws SQLException {
+ List<PFunction> functions = new ArrayList<PFunction>(functionNameAndTimeStampPairs.size());
+ for(Pair<byte[], Long> functionInfo: functionNameAndTimeStampPairs) {
+ try {
+ PFunction function2 = metaData.getFunction(new PTableKey(tenantId, Bytes.toString(functionInfo.getFirst())));
+ functions.add(function2);
+ } catch (FunctionNotFoundException e) {
+ return new MetaDataMutationResult(MutationCode.FUNCTION_NOT_FOUND, 0, null);
+ }
+ }
+ if(functions.isEmpty()) {
+ return null;
+ }
+ return new MetaDataMutationResult(MutationCode.FUNCTION_ALREADY_EXISTS, 0, functions, true);
+ }
+
+ @Override
+ public MetaDataMutationResult dropFunction(List<Mutation> tableMetadata, boolean ifExists)
+ throws SQLException {
+ return new MetaDataMutationResult(MutationCode.FUNCTION_ALREADY_EXISTS, 0, null);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
index 460a879,4153652..1862d53
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/DelegateConnectionQueryServices.java
@@@ -259,7 -254,32 +261,36 @@@ public class DelegateConnectionQuerySer
}
@Override
+ public TransactionSystemClient getTransactionSystemClient() {
+ return getDelegate().getTransactionSystemClient();
+ }
++
+ public MetaDataMutationResult createFunction(List<Mutation> functionData, PFunction function, boolean temporary)
+ throws SQLException {
+ return getDelegate().createFunction(functionData, function, temporary);
+ }
+
+ @Override
+ public PMetaData addFunction(PFunction function) throws SQLException {
+ return getDelegate().addFunction(function);
+ }
+
+ @Override
+ public PMetaData removeFunction(PName tenantId, String function, long functionTimeStamp)
+ throws SQLException {
+ return getDelegate().removeFunction(tenantId, function, functionTimeStamp);
+ }
+
+ @Override
+ public MetaDataMutationResult getFunctions(PName tenantId,
+ List<Pair<byte[], Long>> functionNameAndTimeStampPairs, long clientTimestamp)
+ throws SQLException {
+ return getDelegate().getFunctions(tenantId, functionNameAndTimeStampPairs, clientTimestamp);
+ }
+
+ @Override
+ public MetaDataMutationResult dropFunction(List<Mutation> tableMetadata, boolean ifExists)
+ throws SQLException {
+ return getDelegate().dropFunction(tableMetadata, ifExists);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/MetaDataMutated.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/MetaDataMutated.java
index c78d0b6,76e7593..706abf9
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/MetaDataMutated.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/MetaDataMutated.java
@@@ -34,9 -35,10 +35,11 @@@ import org.apache.phoenix.schema.PTable
* @since 0.1
*/
public interface MetaDataMutated {
- PMetaData addTable(PTable table) throws SQLException;
+ PMetaData addTable(PTable table, long resolvedTime) throws SQLException;
+ PMetaData updateResolvedTimestamp(PTable table, long resolvedTimestamp) throws SQLException;
PMetaData removeTable(PName tenantId, String tableName, String parentTableName, long tableTimeStamp) throws SQLException;
- PMetaData addColumn(PName tenantId, String tableName, List<PColumn> columns, long tableTimeStamp, long tableSeqNum, boolean isImmutableRows, boolean isWalDisabled, boolean isMultitenant, boolean storeNulls) throws SQLException;
- PMetaData removeColumn(PName tenantId, String tableName, List<PColumn> columnsToRemove, long tableTimeStamp, long tableSeqNum) throws SQLException;
+ PMetaData addColumn(PName tenantId, String tableName, List<PColumn> columns, long tableTimeStamp, long tableSeqNum, boolean isImmutableRows, boolean isWalDisabled, boolean isMultitenant, boolean storeNulls, long resolvedTime) throws SQLException;
+ PMetaData removeColumn(PName tenantId, String tableName, List<PColumn> columnsToRemove, long tableTimeStamp, long tableSeqNum, long resolvedTime) throws SQLException;
+ PMetaData addFunction(PFunction function) throws SQLException;
+ PMetaData removeFunction(PName tenantId, String function, long functionTimeStamp) throws SQLException;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 8c2e990,268bfc1..2d42fea
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@@ -84,7 -96,7 +96,8 @@@ import static org.apache.phoenix.jdbc.P
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SEQ_NUM;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTIONAL;
+ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_CONSTANT;
@@@ -233,7 -249,9 +251,10 @@@ public interface QueryConstants
INDEX_TYPE + " UNSIGNED_TINYINT," +
INDEX_DISABLE_TIMESTAMP + " BIGINT," +
STORE_NULLS + " BOOLEAN," +
+ BASE_COLUMN_COUNT + " INTEGER," +
+ // Column metadata (will be null for table row)
- IS_ROW_TIMESTAMP + " BOOLEAN, " +
++ IS_ROW_TIMESTAMP + " BOOLEAN, " +
+ TRANSACTIONAL + " BOOLEAN," +
"CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + ","
+ TABLE_SCHEM + "," + TABLE_NAME + "," + COLUMN_NAME + "," + COLUMN_FAMILY + "))\n" +
HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + ",\n" +
@@@ -280,6 -296,30 +301,31 @@@
LIMIT_REACHED_FLAG + " BOOLEAN \n" +
" CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + "," + SEQUENCE_SCHEMA + "," + SEQUENCE_NAME + "))\n" +
HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + ",\n" +
- HColumnDescriptor.KEEP_DELETED_CELLS + "=" + MetaDataProtocol.DEFAULT_META_DATA_KEEP_DELETED_CELLS + "\n";
+ HColumnDescriptor.KEEP_DELETED_CELLS + "=" + MetaDataProtocol.DEFAULT_META_DATA_KEEP_DELETED_CELLS + ",\n" +
+ PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE;
+
+ public static final String CREATE_FUNCTION_METADATA =
+ "CREATE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_FUNCTION_TABLE + "\"(\n" +
+ // Pk columns
+ TENANT_ID + " VARCHAR NULL," +
+ FUNCTION_NAME + " VARCHAR NOT NULL, \n" +
+ NUM_ARGS + " INTEGER, \n" +
+ // Function metadata (will be null for argument row)
+ CLASS_NAME + " VARCHAR, \n" +
+ JAR_PATH + " VARCHAR, \n" +
+ RETURN_TYPE + " VARCHAR, \n" +
+ // Argument metadata (will be null for function row)
+ TYPE + " VARCHAR, \n" +
+ ARG_POSITION + " VARBINARY, \n" +
+ IS_ARRAY + " BOOLEAN, \n" +
+ IS_CONSTANT + " BOOLEAN, \n" +
+ DEFAULT_VALUE + " VARCHAR, \n" +
+ MIN_VALUE + " VARCHAR, \n" +
+ MAX_VALUE + " VARCHAR, \n" +
+ " CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (" + TENANT_ID + ", " + FUNCTION_NAME + ", " + TYPE + ", " + ARG_POSITION + "))\n" +
+ HConstants.VERSIONS + "=" + MetaDataProtocol.DEFAULT_MAX_META_DATA_VERSIONS + ",\n" +
+ HColumnDescriptor.KEEP_DELETED_CELLS + "=" + MetaDataProtocol.DEFAULT_META_DATA_KEEP_DELETED_CELLS + ",\n"+
+ // Install split policy to prevent a tenant's metadata from being split across regions.
+ HTableDescriptor.SPLIT_POLICY + "='" + MetaDataSplitPolicy.class.getName() + "'\n";
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index 0d2c36f,2e143a6..465bf87
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@@ -152,8 -159,18 +159,19 @@@ public interface QueryServices extends
public static final String DELAY_FOR_SCHEMA_UPDATE_CHECK = "phoenix.schema.change.delay";
public static final String DEFAULT_KEEP_DELETED_CELLS_ATTRIB = "phoenix.table.default.keep.deleted.cells";
public static final String DEFAULT_STORE_NULLS_ATTRIB = "phoenix.table.default.store.nulls";
- public static final String METRICS_ENABLED = "phoenix.query.metrics.enabled";
+ public static final String DEFAULT_TRANSACTIONAL_ATTRIB = "phoenix.transactions.default.enabled";
+ public static final String GLOBAL_METRICS_ENABLED = "phoenix.query.global.metrics.enabled";
+
+ // rpc queue configs
+ public static final String INDEX_HANDLER_COUNT_ATTRIB = "phoenix.rpc.index.handler.count";
+ public static final String METADATA_HANDLER_COUNT_ATTRIB = "phoenix.rpc.metadata.handler.count";
+
+ public static final String FORCE_ROW_KEY_ORDER_ATTRIB = "phoenix.query.force.rowkeyorder";
+ public static final String ALLOW_USER_DEFINED_FUNCTIONS_ATTRIB = "phoenix.functions.allowUserDefinedFunctions";
+ public static final String COLLECT_REQUEST_LEVEL_METRICS = "phoenix.query.request.metrics.enabled";
+ public static final String ALLOW_VIEWS_ADD_NEW_CF_BASE_TABLE = "phoenix.view.allowNewColumnFamily";
+ public static final String RETURN_SEQUENCE_VALUES_ATTRIB = "phoenix.sequence.returnValues";
+
/**
* Get executor service used for parallel scans
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 83944b3,d003f36..c40b6e1
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@@ -181,11 -187,21 +187,23 @@@ public class QueryServicesOptions
public static final boolean DEFAULT_STORE_NULLS = false;
// TODO Change this to true as part of PHOENIX-1543
+ // We'll also need this for transactions to work correctly
public static final boolean DEFAULT_AUTO_COMMIT = false;
- public static final boolean DEFAULT_IS_METRICS_ENABLED = true;
+ public static final boolean DEFAULT_TRANSACTIONAL = false;
+ public static final boolean DEFAULT_IS_GLOBAL_METRICS_ENABLED = true;
+
+ private static final String DEFAULT_CLIENT_RPC_CONTROLLER_FACTORY = ClientRpcControllerFactory.class.getName();
+
+ public static final String DEFAULT_CONSISTENCY_LEVEL = Consistency.STRONG.toString();
+
+ public static final boolean DEFAULT_USE_BYTE_BASED_REGEX = false;
+ public static final boolean DEFAULT_FORCE_ROW_KEY_ORDER = false;
+ public static final boolean DEFAULT_ALLOW_USER_DEFINED_FUNCTIONS = false;
+ public static final boolean DEFAULT_REQUEST_LEVEL_METRICS_ENABLED = false;
+ public static final boolean DEFAULT_ALLOW_VIEWS_ADD_NEW_CF_BASE_TABLE = true;
+ public static final boolean DEFAULT_RETURN_SEQUENCE_VALUES = false;
+
private final Configuration config;
private QueryServicesOptions(Configuration config) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
index 98f9f4a,5a0d63b..7fb90a1
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/DelegateTable.java
@@@ -238,7 -238,22 +238,26 @@@ public class DelegateTable implements P
}
@Override
+ public boolean isTransactional() {
+ return delegate.isTransactional();
+ }
++
+ public int getBaseColumnCount() {
+ return delegate.getBaseColumnCount();
+ }
+
+ @Override
+ public boolean rowKeyOrderOptimizable() {
+ return delegate.rowKeyOrderOptimizable();
+ }
+
+ @Override
+ public int getRowTimestampColPos() {
+ return delegate.getRowTimestampColPos();
+ }
+
+ @Override
+ public String toString() {
+ return delegate.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index b2982e4,77ca2c3..f69c44c
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@@ -58,7 -72,7 +72,8 @@@ import static org.apache.phoenix.jdbc.P
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SEQ_NUM;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTIONAL;
+ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_CONSTANT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_INDEX_ID;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT;
@@@ -97,11 -117,10 +117,10 @@@ import org.apache.hadoop.hbase.client.M
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
- import org.apache.hadoop.hbase.io.TimeRange;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.compile.BaseMutationPlan;
import org.apache.phoenix.compile.ColumnResolver;
-import org.apache.phoenix.compile.ExplainPlan;
import org.apache.phoenix.compile.FromCompiler;
import org.apache.phoenix.compile.IndexExpressionCompiler;
import org.apache.phoenix.compile.MutationPlan;
@@@ -142,9 -166,10 +166,11 @@@ import org.apache.phoenix.parse.PFuncti
import org.apache.phoenix.parse.ParseNode;
import org.apache.phoenix.parse.ParseNodeFactory;
import org.apache.phoenix.parse.PrimaryKeyConstraint;
+ import org.apache.phoenix.parse.SQLParser;
+ import org.apache.phoenix.parse.SelectStatement;
import org.apache.phoenix.parse.TableName;
import org.apache.phoenix.parse.UpdateStatisticsStatement;
+import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.ConnectionQueryServices.Feature;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
@@@ -163,8 -191,10 +192,11 @@@ import org.apache.phoenix.util.LogUtil
import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.ReadOnlyProps;
+ import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.TransactionUtil;
+ import org.apache.phoenix.util.StringUtil;
+ import org.apache.phoenix.util.UpgradeUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -201,8 -231,8 +233,9 @@@ public class MetaDataClient
VIEW_INDEX_ID + "," +
INDEX_TYPE + "," +
STORE_NULLS + "," +
- BASE_COLUMN_COUNT +
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
++ BASE_COLUMN_COUNT + "," +
+ TRANSACTIONAL +
- ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
++ ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
private static final String CREATE_LINK =
"UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
TENANT_ID + "," +
@@@ -315,19 -392,34 +395,43 @@@
public MetaDataMutationResult updateCache(PName tenantId, String schemaName, String tableName) throws SQLException {
return updateCache(tenantId, schemaName, tableName, false);
}
+
+ public MetaDataMutationResult updateCache(PName tenantId, String schemaName, String tableName, boolean alwaysHitServer) throws SQLException {
+ return updateCache(tenantId, schemaName, tableName, alwaysHitServer, null);
+ }
+
+ /**
+ * Update the cache with the latest as of the connection scn.
+ * @param functioNames
+ * @return the timestamp from the server, negative if the function was added to the cache and positive otherwise
+ * @throws SQLException
+ */
+ public MetaDataMutationResult updateCache(List<String> functionNames) throws SQLException {
+ return updateCache(functionNames, false);
+ }
+
+ private MetaDataMutationResult updateCache(List<String> functionNames, boolean alwaysHitServer) throws SQLException {
+ return updateCache(connection.getTenantId(), functionNames, alwaysHitServer);
+ }
+
+ public MetaDataMutationResult updateCache(PName tenantId, List<String> functionNames) throws SQLException {
+ return updateCache(tenantId, functionNames, false);
+ }
+
+ private long getClientTimeStamp() {
+ Long scn = connection.getSCN();
+ long clientTimeStamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
+ return clientTimeStamp;
+ }
-
+
+ private long getCurrentScn() {
+ Long scn = connection.getSCN();
+ long currentScn = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
+ return currentScn;
+ }
+
private MetaDataMutationResult updateCache(PName tenantId, String schemaName, String tableName,
- boolean alwaysHitServer) throws SQLException { // TODO: pass byte[] herez
- long clientTimeStamp = getClientTimeStamp();
+ boolean alwaysHitServer, Long resolvedTimestamp) throws SQLException { // TODO: pass byte[] herez
boolean systemTable = SYSTEM_CATALOG_SCHEMA.equals(schemaName);
// System tables must always have a null tenantId
tenantId = systemTable ? null : tenantId;
@@@ -795,70 -963,86 +1004,66 @@@
// the wire for the index rows, as we don't need to do that. Instead, we tap into our
// region observer to generate the index rows based on the data rows as we scan
if (index.getIndexType() == IndexType.LOCAL) {
- final PhoenixStatement statement = new PhoenixStatement(connection);
- String tableName = getFullTableName(dataTableRef);
- String query = "SELECT count(*) FROM " + tableName;
- final QueryPlan plan = statement.compileQuery(query);
- TableRef tableRef = plan.getTableRef();
- // Set attribute on scan that UngroupedAggregateRegionObserver will switch on.
- // We'll detect that this attribute was set the server-side and write the index
- // rows per region as a result. The value of the attribute will be our persisted
- // index maintainers.
- // Define the LOCAL_INDEX_BUILD as a new static in BaseScannerRegionObserver
- Scan scan = plan.getContext().getScan();
- try {
- if(plan.getContext().getScanTimeRange()==null) {
- Long scn = connection.getSCN();
- if (scn == null) {
- scn = plan.getContext().getCurrentTime();
- // Add one to server time since max of time range is exclusive
- // and we need to account of OSs with lower resolution clocks.
- if (scn < HConstants.LATEST_TIMESTAMP) {
- scn++;
+ try (final PhoenixStatement statement = new PhoenixStatement(connection)) {
+ String tableName = getFullTableName(dataTableRef);
+ String query = "SELECT count(*) FROM " + tableName;
+ final QueryPlan plan = statement.compileQuery(query);
+ TableRef tableRef = plan.getTableRef();
+ // Set attribute on scan that UngroupedAggregateRegionObserver will switch on.
+ // We'll detect that this attribute was set the server-side and write the index
+ // rows per region as a result. The value of the attribute will be our persisted
+ // index maintainers.
+ // Define the LOCAL_INDEX_BUILD as a new static in BaseScannerRegionObserver
+ Scan scan = plan.getContext().getScan();
+ try {
+ if(ScanUtil.isDefaultTimeRange(scan.getTimeRange())) {
+ Long scn = connection.getSCN();
+ if (scn == null) {
+ scn = plan.getContext().getCurrentTime();
}
+ scan.setTimeRange(dataTableRef.getLowerBoundTimeStamp(),scn);
+ }
+ } catch (IOException e) {
+ throw new SQLException(e);
+ }
+ ImmutableBytesWritable ptr = new ImmutableBytesWritable();
+ PTable dataTable = tableRef.getTable();
+ for(PTable idx: dataTable.getIndexes()) {
+ if(idx.getName().equals(index.getName())) {
+ index = idx;
+ break;
}
- plan.getContext().setScanTimeRange(new TimeRange(dataTableRef.getLowerBoundTimeStamp(),scn));
}
- } catch (IOException e) {
- throw new SQLException(e);
- }
- ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- PTable dataTable = tableRef.getTable();
- for(PTable idx: dataTable.getIndexes()) {
- if(idx.getName().equals(index.getName())) {
- index = idx;
- break;
+ List<PTable> indexes = Lists.newArrayListWithExpectedSize(1);
+ // Only build newly created index.
+ indexes.add(index);
+ IndexMaintainer.serialize(dataTable, ptr, indexes, plan.getContext().getConnection());
+ scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD, ByteUtil.copyKeyBytesIfNecessary(ptr));
+ // By default, we'd use a FirstKeyOnly filter as nothing else needs to be projected for count(*).
+ // However, in this case, we need to project all of the data columns that contribute to the index.
+ IndexMaintainer indexMaintainer = index.getIndexMaintainer(dataTable, connection);
+ for (ColumnReference columnRef : indexMaintainer.getAllColumns()) {
+ scan.addColumn(columnRef.getFamily(), columnRef.getQualifier());
}
- }
- List<PTable> indexes = Lists.newArrayListWithExpectedSize(1);
- // Only build newly created index.
- indexes.add(index);
- IndexMaintainer.serialize(dataTable, ptr, indexes, plan.getContext().getConnection());
- scan.setAttribute(BaseScannerRegionObserver.LOCAL_INDEX_BUILD, ByteUtil.copyKeyBytesIfNecessary(ptr));
- // By default, we'd use a FirstKeyOnly filter as nothing else needs to be projected for count(*).
- // However, in this case, we need to project all of the data columns that contribute to the index.
- IndexMaintainer indexMaintainer = index.getIndexMaintainer(dataTable, connection);
- for (ColumnReference columnRef : indexMaintainer.getAllColumns()) {
- scan.addColumn(columnRef.getFamily(), columnRef.getQualifier());
- }
- // Go through MutationPlan abstraction so that we can create local indexes
- // with a connectionless connection (which makes testing easier).
- mutationPlan = new MutationPlan() {
-
- @Override
- public StatementContext getContext() {
- return plan.getContext();
- }
-
- @Override
- public ParameterMetaData getParameterMetaData() {
- return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA;
- }
-
- @Override
- public ExplainPlan getExplainPlan() throws SQLException {
- return ExplainPlan.EMPTY_PLAN;
- }
-
- @Override
- public PhoenixConnection getConnection() {
- return connection;
- }
+ // Go through MutationPlan abstraction so that we can create local indexes
+ // with a connectionless connection (which makes testing easier).
+ mutationPlan = new BaseMutationPlan(plan.getContext(), Operation.UPSERT) {
- @Override
- public MutationState execute() throws SQLException {
- Cell kv = plan.iterator().next().getValue(0);
- ImmutableBytesWritable tmpPtr = new ImmutableBytesWritable(kv.getValueArray(), kv.getValueOffset(), kv.getValueLength());
- // A single Cell will be returned with the count(*) - we decode that here
- long rowCount = PLong.INSTANCE.getCodec().decodeLong(tmpPtr, SortOrder.getDefault());
- // The contract is to return a MutationState that contains the number of rows modified. In this
- // case, it's the number of rows in the data table which corresponds to the number of index
- // rows that were added.
- return new MutationState(0, connection, rowCount);
- }
+ @Override
+ public MutationState execute() throws SQLException {
+ Cell kv = plan.iterator().next().getValue(0);
+ ImmutableBytesWritable tmpPtr = new ImmutableBytesWritable(kv.getValueArray(), kv.getValueOffset(), kv.getValueLength());
+ // A single Cell will be returned with the count(*) - we decode that here
+ long rowCount = PLong.INSTANCE.getCodec().decodeLong(tmpPtr, SortOrder.getDefault());
+ // The contract is to return a MutationState that contains the number of rows modified. In this
+ // case, it's the number of rows in the data table which corresponds to the number of index
+ // rows that were added.
+ return new MutationState(0, connection, rowCount);
+ }
- };
+ };
+ }
} else {
PostIndexDDLCompiler compiler = new PostIndexDDLCompiler(connection, dataTableRef);
mutationPlan = compiler.compile(index);
@@@ -1103,15 -1295,15 +1315,16 @@@
PName physicalName = dataTable.getPhysicalName();
int nSequenceSaltBuckets = connection.getQueryServices().getSequenceSaltBuckets();
SequenceKey key = MetaDataUtil.getViewIndexSequenceKey(tenantIdStr, physicalName, nSequenceSaltBuckets);
- // Create at parent timestamp as we know that will be earlier than now
- // and earlier than any SCN if one is set.
+ // if scn is set create at scn-1, so we can see the sequence or else use latest timestamp (so that latest server time is used)
+ long sequenceTimestamp = scn!=null ? scn-1 : HConstants.LATEST_TIMESTAMP;
createSequence(key.getTenantId(), key.getSchemaName(), key.getSequenceName(),
true, Short.MIN_VALUE, 1, 1, false, Long.MIN_VALUE, Long.MAX_VALUE,
- dataTable.getTimeStamp());
+ sequenceTimestamp);
long[] seqValues = new long[1];
SQLException[] sqlExceptions = new SQLException[1];
+ long timestamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
- connection.getQueryServices().incrementSequences(Collections.singletonList(key), timestamp, seqValues, sqlExceptions);
+ connection.getQueryServices().incrementSequences(Collections.singletonList(new SequenceAllocation(key, 1)),
+ Math.max(timestamp, dataTable.getTimeStamp()), seqValues, sqlExceptions);
if (sqlExceptions[0] != null) {
throw sqlExceptions[0];
}
@@@ -1194,6 -1391,67 +1412,67 @@@
return new MutationState(1, connection);
}
+ public MutationState createFunction(CreateFunctionStatement stmt) throws SQLException {
+ boolean wasAutoCommit = connection.getAutoCommit();
+ connection.rollback();
+ try {
+ PFunction function = new PFunction(stmt.getFunctionInfo(), stmt.isTemporary(), stmt.isReplace());
+ connection.setAutoCommit(false);
+ String tenantIdStr = connection.getTenantId() == null ? null : connection.getTenantId().getString();
+ List<Mutation> functionData = Lists.newArrayListWithExpectedSize(function.getFunctionArguments().size() + 1);
+
+ List<FunctionArgument> args = function.getFunctionArguments();
+ PreparedStatement argUpsert = connection.prepareStatement(INSERT_FUNCTION_ARGUMENT);
+
+ for (int i = 0; i < args.size(); i++) {
+ FunctionArgument arg = args.get(i);
+ addFunctionArgMutation(function.getFunctionName(), arg, argUpsert, i);
+ }
- functionData.addAll(connection.getMutationState().toMutations().next().getSecond());
++ functionData.addAll(connection.getMutationState().toMutations(null).next().getSecond());
+ connection.rollback();
+
+ PreparedStatement functionUpsert = connection.prepareStatement(CREATE_FUNCTION);
+ functionUpsert.setString(1, tenantIdStr);
+ functionUpsert.setString(2, function.getFunctionName());
+ functionUpsert.setInt(3, function.getFunctionArguments().size());
+ functionUpsert.setString(4, function.getClassName());
+ functionUpsert.setString(5, function.getJarPath());
+ functionUpsert.setString(6, function.getReturnType());
+ functionUpsert.execute();
- functionData.addAll(connection.getMutationState().toMutations().next().getSecond());
++ functionData.addAll(connection.getMutationState().toMutations(null).next().getSecond());
+ connection.rollback();
+ MetaDataMutationResult result = connection.getQueryServices().createFunction(functionData, function, stmt.isTemporary());
+ MutationCode code = result.getMutationCode();
+ switch(code) {
+ case FUNCTION_ALREADY_EXISTS:
+ if (!function.isReplace()) {
+ throw new FunctionAlreadyExistsException(function.getFunctionName(), result
+ .getFunctions().get(0));
+ } else {
+ connection.removeFunction(function.getTenantId(), function.getFunctionName(),
+ result.getMutationTime());
+ addFunctionToCache(result);
+ }
+ case NEWER_FUNCTION_FOUND:
+ // Add function to ConnectionQueryServices so it's cached, but don't add
+ // it to this connection as we can't see it.
+ throw new NewerFunctionAlreadyExistsException(function.getFunctionName(), result.getFunctions().get(0));
+ default:
+ List<PFunction> functions = new ArrayList<PFunction>(1);
+ functions.add(function);
+ result = new MetaDataMutationResult(code, result.getMutationTime(), functions, true);
+ if(function.isReplace()) {
+ connection.removeFunction(function.getTenantId(), function.getFunctionName(),
+ result.getMutationTime());
+ }
+ addFunctionToCache(result);
+ }
+ } finally {
+ connection.setAutoCommit(wasAutoCommit);
+ }
+ return new MutationState(1, connection);
+ }
+
private static ColumnDef findColumnDefOrNull(List<ColumnDef> colDefs, ColumnName colName) {
for (ColumnDef colDef : colDefs) {
if (colDef.getColumnDefName().getColumnName().equals(colName.getColumnName())) {
@@@ -1216,63 -1514,61 +1535,92 @@@
String tableName = tableNameNode.getTableName();
String parentTableName = null;
PName tenantId = connection.getTenantId();
- String tenantIdStr = tenantId == null ? null : connection.getTenantId().getString();
+ String tenantIdStr = tenantId == null ? null : tenantId.getString();
+ Long scn = connection.getSCN();
+ long clientTimeStamp = scn == null ? HConstants.LATEST_TIMESTAMP : scn;
boolean multiTenant = false;
boolean storeNulls = false;
+ boolean transactional = false;
Integer saltBucketNum = null;
String defaultFamilyName = null;
boolean isImmutableRows = false;
List<PName> physicalNames = Collections.emptyList();
boolean addSaltColumn = false;
- Long timestamp = null;
- if (parent != null) {
- transactional = parent.isTransactional();
+ boolean rowKeyOrderOptimizable = true;
++ Long timestamp = null;
+ if (parent != null && tableType == PTableType.INDEX) {
- // Index on view
- // TODO: Can we support a multi-tenant index directly on a multi-tenant
- // table instead of only a view? We don't have anywhere to put the link
- // from the table to the index, though.
- if (indexType == IndexType.LOCAL || (parent.getType() == PTableType.VIEW && parent.getViewType() != ViewType.MAPPED)) {
- PName physicalName = parent.getPhysicalName();
- saltBucketNum = parent.getBucketNum();
- addSaltColumn = (saltBucketNum != null && indexType != IndexType.LOCAL);
- defaultFamilyName = parent.getDefaultFamilyName() == null ? null : parent.getDefaultFamilyName().getString();
- if (indexType == IndexType.LOCAL) {
- saltBucketNum = null;
- // Set physical name of local index table
- physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getLocalIndexPhysicalName(physicalName.getBytes())));
- } else {
- // Set physical name of view index table
- physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getViewIndexPhysicalName(physicalName.getBytes())));
++ transactional = parent.isTransactional();
+ timestamp = TransactionUtil.getTableTimestamp(connection, transactional);
+ storeNulls = parent.getStoreNulls();
+ if (tableType == PTableType.INDEX) {
- // Index on view
- // TODO: Can we support a multi-tenant index directly on a multi-tenant
- // table instead of only a view? We don't have anywhere to put the link
- // from the table to the index, though.
- if (indexType == IndexType.LOCAL || (parent.getType() == PTableType.VIEW && parent.getViewType() != ViewType.MAPPED)) {
- PName physicalName = parent.getPhysicalName();
- saltBucketNum = parent.getBucketNum();
- addSaltColumn = (saltBucketNum != null && indexType != IndexType.LOCAL);
- defaultFamilyName = parent.getDefaultFamilyName() == null ? null : parent.getDefaultFamilyName().getString();
- if (indexType == IndexType.LOCAL) {
- saltBucketNum = null;
- // Set physical name of local index table
- physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getLocalIndexPhysicalName(physicalName.getBytes())));
- } else {
- // Set physical name of view index table
- physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getViewIndexPhysicalName(physicalName.getBytes())));
- }
++ // Index on view
++ // TODO: Can we support a multi-tenant index directly on a multi-tenant
++ // table instead of only a view? We don't have anywhere to put the link
++ // from the table to the index, though.
++ if (indexType == IndexType.LOCAL || (parent.getType() == PTableType.VIEW && parent.getViewType() != ViewType.MAPPED)) {
++ PName physicalName = parent.getPhysicalName();
++ saltBucketNum = parent.getBucketNum();
++ addSaltColumn = (saltBucketNum != null && indexType != IndexType.LOCAL);
++ defaultFamilyName = parent.getDefaultFamilyName() == null ? null : parent.getDefaultFamilyName().getString();
++ if (indexType == IndexType.LOCAL) {
++ saltBucketNum = null;
++ // Set physical name of local index table
++ physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getLocalIndexPhysicalName(physicalName.getBytes())));
++ } else {
++ // Set physical name of view index table
++ physicalNames = Collections.singletonList(PNameFactory.newName(MetaDataUtil.getViewIndexPhysicalName(physicalName.getBytes())));
++ }
}
+
+ multiTenant = parent.isMultiTenant();
+ parentTableName = parent.getTableName().getString();
+ // Pass through data table sequence number so we can check it hasn't changed
+ PreparedStatement incrementStatement = connection.prepareStatement(INCREMENT_SEQ_NUM);
+ incrementStatement.setString(1, connection.getTenantId() == null ? null : connection.getTenantId().getString());
+ incrementStatement.setString(2, schemaName);
+ incrementStatement.setString(3, parentTableName);
+ incrementStatement.setLong(4, parent.getSequenceNumber());
+ incrementStatement.execute();
+ // Get list of mutations and add to table meta data that will be passed to server
+ // to guarantee order. This row will always end up last
+ tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
+ connection.rollback();
+
+ // Add row linking from data table row to index table row
+ PreparedStatement linkStatement = connection.prepareStatement(CREATE_LINK);
+ linkStatement.setString(1, connection.getTenantId() == null ? null : connection.getTenantId().getString());
+ linkStatement.setString(2, schemaName);
+ linkStatement.setString(3, parentTableName);
+ linkStatement.setString(4, tableName);
+ linkStatement.setByte(5, LinkType.INDEX_TABLE.getSerializedValue());
++ linkStatement.setLong(6, parent.getSequenceNumber());
+ linkStatement.execute();
}
+
+ multiTenant = parent.isMultiTenant();
+ storeNulls = parent.getStoreNulls();
+ parentTableName = parent.getTableName().getString();
+ // Pass through data table sequence number so we can check it hasn't changed
+ PreparedStatement incrementStatement = connection.prepareStatement(INCREMENT_SEQ_NUM);
+ incrementStatement.setString(1, tenantIdStr);
+ incrementStatement.setString(2, schemaName);
+ incrementStatement.setString(3, parentTableName);
+ incrementStatement.setLong(4, parent.getSequenceNumber());
+ incrementStatement.execute();
+ // Get list of mutations and add to table meta data that will be passed to server
+ // to guarantee order. This row will always end up last
- tableMetaData.addAll(connection.getMutationState().toMutations().next().getSecond());
++ tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
+ connection.rollback();
+
+ // Add row linking from data table row to index table row
+ PreparedStatement linkStatement = connection.prepareStatement(CREATE_LINK);
+ linkStatement.setString(1, tenantIdStr);
+ linkStatement.setString(2, schemaName);
+ linkStatement.setString(3, parentTableName);
+ linkStatement.setString(4, tableName);
+ linkStatement.setByte(5, LinkType.INDEX_TABLE.getSerializedValue());
+ linkStatement.setLong(6, parent.getSequenceNumber());
+ linkStatement.execute();
}
PrimaryKeyConstraint pkConstraint = statement.getPrimaryKeyConstraint();
@@@ -1461,6 -1727,12 +1815,12 @@@
linkStatement.setString(3, tableName);
linkStatement.setString(4, physicalName.getString());
linkStatement.setByte(5, LinkType.PHYSICAL_TABLE.getSerializedValue());
+ if (tableType == PTableType.VIEW) {
- PTable physicalTable = connection.getMetaDataCache().getTable(new PTableKey(null, physicalName.getString()));
++ PTable physicalTable = connection.getMetaDataCache().getTableRef(new PTableKey(null, physicalName.getString())).getTable();
+ linkStatement.setLong(6, physicalTable.getSequenceNumber());
+ } else {
+ linkStatement.setLong(6, parent.getSequenceNumber());
+ }
linkStatement.execute();
}
}
@@@ -1618,8 -1887,8 +1975,8 @@@
Collections.<PTable>emptyList(), isImmutableRows,
Collections.<PName>emptyList(), defaultFamilyName == null ? null :
PNameFactory.newName(defaultFamilyName), null,
- Boolean.TRUE.equals(disableWAL), false, false, null, indexId, indexType, false);
- Boolean.TRUE.equals(disableWAL), false, false, null, indexId, indexType, true);
- connection.addTable(table);
++ Boolean.TRUE.equals(disableWAL), false, false, null, indexId, indexType, true, false);
+ connection.addTable(table, MetaDataProtocol.MIN_TABLE_TIMESTAMP);
} else if (tableType == PTableType.INDEX && indexId == null) {
if (tableProps.get(HTableDescriptor.MAX_FILESIZE) == null) {
int nIndexRowKeyColumns = isPK ? 1 : pkColumnsNames.size();
@@@ -1715,10 -1984,14 +2072,15 @@@
tableUpsert.setByte(18, indexType.getSerializedValue());
}
tableUpsert.setBoolean(19, storeNulls);
- tableUpsert.setBoolean(20, transactional);
+ if (parent != null && tableType == PTableType.VIEW) {
+ tableUpsert.setInt(20, parent.getColumns().size());
+ } else {
+ tableUpsert.setInt(20, BASE_TABLE_BASE_COLUMN_COUNT);
+ }
++ tableUpsert.setBoolean(21, transactional);
tableUpsert.execute();
- tableMetaData.addAll(connection.getMutationState().toMutations().next().getSecond());
+ tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
connection.rollback();
/*
@@@ -1770,7 -2041,7 +2132,7 @@@
PTable.INITIAL_SEQ_NUM, pkName == null ? null : PNameFactory.newName(pkName), saltBucketNum, columns,
dataTableName == null ? null : newSchemaName, dataTableName == null ? null : PNameFactory.newName(dataTableName), Collections.<PTable>emptyList(), isImmutableRows,
physicalNames, defaultFamilyName == null ? null : PNameFactory.newName(defaultFamilyName), viewStatement, Boolean.TRUE.equals(disableWAL), multiTenant, storeNulls, viewType,
- indexId, indexType, transactional);
- indexId, indexType, rowKeyOrderOptimizable);
++ indexId, indexType, rowKeyOrderOptimizable, transactional);
result = new MetaDataMutationResult(code, result.getMutationTime(), table, true);
addTableToCache(result);
return table;
@@@ -2257,7 -2579,7 +2672,7 @@@
}
}
- tableMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond());
- columnMetaData.addAll(connection.getMutationState().toMutations().next().getSecond());
++ columnMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond());
connection.rollback();
} else {
// Check that HBase configured properly for mutable secondary indexing
@@@ -2286,8 -2608,8 +2701,8 @@@
}
long seqNum = table.getSequenceNumber();
if (changingPhoenixTableProperty || columnDefs.size() > 0) {
- seqNum = incrementTableSeqNum(table, statement.getTableType(), 1, isImmutableRows, disableWAL, multiTenant, storeNulls);
+ seqNum = incrementTableSeqNum(table, statement.getTableType(), columnDefs.size(), isImmutableRows, disableWAL, multiTenant, storeNulls);
- tableMetaData.addAll(connection.getMutationState().toMutations().next().getSecond());
+ tableMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond());
connection.rollback();
}
@@@ -2709,20 -3024,17 +3128,17 @@@
}
private PTable addTableToCache(MetaDataMutationResult result) throws SQLException {
- addIndexesFromPhysicalTable(result);
+ addIndexesFromPhysicalTable(result, null);
PTable table = result.getTable();
- connection.addTable(table);
+ connection.addTable(table, TransactionUtil.getResolvedTime(connection, result));
return table;
}
-
- private void throwIfAlteringViewPK(ColumnDef col, PTable table) throws SQLException {
- if (col != null && col.isPK() && table.getType() == PTableType.VIEW) {
- throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_MODIFY_VIEW_PK)
- .setSchemaName(table.getSchemaName().getString())
- .setTableName(table.getTableName().getString())
- .setColumnName(col.getColumnDefName().getColumnName())
- .build().buildException();
+
+ private List<PFunction> addFunctionToCache(MetaDataMutationResult result) throws SQLException {
+ for(PFunction function: result.getFunctions()) {
+ connection.addFunction(function);
}
+ return result.getFunctions();
}
public PTableStats getTableStats(PTable table) throws SQLException {
@@@ -2760,4 -3071,24 +3176,24 @@@
return table.getTableStats();
}
+ private void throwIfLastPKOfParentIsFixedLength(PTable parent, String viewSchemaName, String viewName, ColumnDef col) throws SQLException {
+ if (isLastPKVariableLength(parent)) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_MODIFY_VIEW_PK)
+ .setSchemaName(viewSchemaName)
+ .setTableName(viewName)
+ .setColumnName(col.getColumnDefName().getColumnName())
+ .build().buildException(); }
+ }
+
+ private boolean isLastPKVariableLength(PTable table) {
+ List<PColumn> pkColumns = table.getPKColumns();
+ return !pkColumns.get(pkColumns.size()-1).getDataType().isFixedWidth();
+ }
+
+ private PTable getParentOfView(PTable view) throws SQLException {
+ //TODO just use view.getParentName().getString() after implementing https://issues.apache.org/jira/browse/PHOENIX-2114
+ SelectStatement select = new SQLParser(view.getViewStatement()).parseQuery();
+ String parentName = SchemaUtil.normalizeFullTableName(select.getFrom().toString().trim());
- return connection.getMetaDataCache().getTable(new PTableKey(view.getTenantId(), parentName));
++ return connection.getMetaDataCache().getTableRef(new PTableKey(view.getTenantId(), parentName)).getTable();
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaData.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaData.java
index 207bc2a,f015177..a3103bf
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaData.java
@@@ -17,17 -17,19 +17,21 @@@
*/
package org.apache.phoenix.schema;
+import java.sql.SQLException;
+
+ import org.apache.phoenix.parse.PFunction;
import org.apache.phoenix.query.MetaDataMutated;
-
public interface PMetaData extends MetaDataMutated, Iterable<PTable>, Cloneable {
+
public static interface Pruner {
public boolean prune(PTable table);
+ public boolean prune(PFunction function);
}
public int size();
public PMetaData clone();
- public PTable getTable(PTableKey key) throws TableNotFoundException;
+ public PTableRef getTableRef(PTableKey key) throws TableNotFoundException;
public PMetaData pruneTables(Pruner pruner);
+ public PFunction getFunction(PTableKey key) throws FunctionNotFoundException;
+ public PMetaData pruneFunctions(Pruner pruner);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaDataImpl.java
----------------------------------------------------------------------
diff --cc phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaDataImpl.java
index 8cfbb18,a15de13..baea44c
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaDataImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/PMetaDataImpl.java
@@@ -40,7 -41,27 +41,7 @@@ import com.google.common.primitives.Lon
*
*/
public class PMetaDataImpl implements PMetaData {
- private static class PTableCache implements Cloneable {
- private static final class PTableRef {
- public final PTable table;
- public final int estSize;
- public volatile long lastAccessTime;
-
- public PTableRef(PTable table, long lastAccessTime, int estSize) {
- this.table = table;
- this.lastAccessTime = lastAccessTime;
- this.estSize = estSize;
- }
-
- public PTableRef(PTable table, long lastAccessTime) {
- this (table, lastAccessTime, table.getEstimatedSize());
- }
-
- public PTableRef(PTableRef tableRef) {
- this (tableRef.table, tableRef.lastAccessTime, tableRef.estSize);
- }
- }
-
+ private static class PMetaDataCache implements Cloneable {
private static final int MIN_REMOVAL_SIZE = 3;
private static final Comparator<PTableRef> COMPARATOR = new Comparator<PTableRef>() {
@Override
@@@ -117,13 -156,13 +136,13 @@@
long toRemoveBytes = 0;
// Add to new cache, but track references to remove when done
// to bring cache at least overage amount below it's max size.
- for (PTableRef tableRef : tables.values()) {
+ for (PTableRef tableRef : this.tables.values()) {
- newCache.put(tableRef.table.getKey(), new PTableRef(tableRef));
+ newCache.put(tableRef.getTable().getKey(), new PTableRef(tableRef));
toRemove.add(tableRef);
- toRemoveBytes += tableRef.estSize;
- while (toRemoveBytes - toRemove.peekLast().estSize >= overage) {
+ toRemoveBytes += tableRef.getEstSize();
- if (toRemoveBytes - toRemove.peekLast().getEstSize() > overage) {
++ while (toRemoveBytes - toRemove.peekLast().getEstSize() >= overage) {
PTableRef removedRef = toRemove.removeLast();
- toRemoveBytes -= removedRef.estSize;
+ toRemoveBytes -= removedRef.getEstSize();
}
}
for (PTableRef toRemoveRef : toRemove) {
@@@ -133,12 -172,12 +152,12 @@@
}
private PTable put(PTableKey key, PTableRef ref) {
- currentByteSize += ref.estSize;
+ currentByteSize += ref.getEstSize();
- PTableRef oldTableAccess = tables.put(key, ref);
+ PTableRef oldTableAccess = this.tables.put(key, ref);
PTable oldTable = null;
if (oldTableAccess != null) {
- currentByteSize -= oldTableAccess.estSize;
- oldTable = oldTableAccess.table;
+ currentByteSize -= oldTableAccess.getEstSize();
+ oldTable = oldTableAccess.getTable();
}
return oldTable;
}
@@@ -228,15 -276,9 +256,15 @@@
return metaData.size();
}
+ @Override
+ public PMetaData updateResolvedTimestamp(PTable table, long resolvedTimestamp) throws SQLException {
- PTableCache clone = metaData.clone();
++ PMetaDataCache clone = metaData.clone();
+ clone.putDuplicate(table.getKey(), table, resolvedTimestamp);
+ return new PMetaDataImpl(clone);
+ }
@Override
- public PMetaData addTable(PTable table) throws SQLException {
+ public PMetaData addTable(PTable table, long resolvedTime) throws SQLException {
int netGain = 0;
PTableKey key = table.getKey();
PTableRef oldTableRef = metaData.get(key);
@@@ -270,18 -311,18 +298,18 @@@
netGain += table.getEstimatedSize();
}
long overage = metaData.getCurrentSize() + netGain - metaData.getMaxSize();
- PTableCache tables = overage <= 0 ? metaData.clone() : metaData.cloneMinusOverage(overage);
+ PMetaDataCache newMetaData = overage <= 0 ? metaData.clone() : metaData.cloneMinusOverage(overage);
if (newParentTable != null) { // Upsert new index table into parent data table list
- tables.put(newParentTable.getKey(), newParentTable, parentResolvedTimestamp);
- tables.putDuplicate(table.getKey(), table, resolvedTime);
- newMetaData.put(newParentTable.getKey(), newParentTable);
- newMetaData.putDuplicate(table.getKey(), table);
++ newMetaData.put(newParentTable.getKey(), newParentTable, parentResolvedTimestamp);
++ newMetaData.putDuplicate(table.getKey(), table, resolvedTime);
} else {
- tables.put(table.getKey(), table, resolvedTime);
- newMetaData.put(table.getKey(), table);
++ newMetaData.put(table.getKey(), table, resolvedTime);
}
for (PTable index : table.getIndexes()) {
- tables.putDuplicate(index.getKey(), index, resolvedTime);
- newMetaData.putDuplicate(index.getKey(), index);
++ newMetaData.putDuplicate(index.getKey(), index, resolvedTime);
}
- return new PMetaDataImpl(tables);
+ return new PMetaDataImpl(newMetaData);
}
@Override
@@@ -357,8 -398,8 +385,8 @@@
if (tableRef == null) {
return this;
}
- PTable table = tableRef.table;
+ PTable table = tableRef.getTable();
- PTableCache tables = metaData.clone();
+ PMetaDataCache tables = metaData.clone();
for (PColumn columnToRemove : columnsToRemove) {
PColumn column;
String familyName = columnToRemove.getFamilyName().getString();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7d3b544a/phoenix-core/src/main/java/org/apache/phoenix/schema/PTable.java
----------------------------------------------------------------------