PHOENIX-2589 Fix a few resource leaks, NULL dereference, NULL_RETURNS issues (Samarth Jain, Alicia Ying Shu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6911770e Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6911770e Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6911770e Branch: refs/heads/calcite Commit: 6911770e8a1d05775f3780f623ae01e9122d59f0 Parents: 6ecbbb2 Author: Samarth <samarth.j...@salesforce.com> Authored: Wed Jan 20 17:07:12 2016 -0800 Committer: Samarth <samarth.j...@salesforce.com> Committed: Wed Jan 20 17:07:12 2016 -0800 ---------------------------------------------------------------------- .../query/ConnectionQueryServicesImpl.java | 20 +- .../apache/phoenix/schema/MetaDataClient.java | 235 ++++++++++--------- pom.xml | 1 + 3 files changed, 137 insertions(+), 119 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/6911770e/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java index a246e63..4522cf8 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java @@ -3311,8 +3311,14 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement wait = false; } // It is guaranteed that this poll won't hang indefinitely because this is the - // only thread that removes items from the queue. - WeakReference<PhoenixConnection> connRef = connectionsQueue.poll(); + // only thread that removes items from the queue. Still adding a 1 ms timeout + // for sanity check. + WeakReference<PhoenixConnection> connRef = + connectionsQueue.poll(1, TimeUnit.MILLISECONDS); + if (connRef == null) { + throw new IllegalStateException( + "Connection ref found to be null. This is a bug. Some other thread removed items from the connection queue."); + } PhoenixConnection conn = connRef.get(); if (conn != null && !conn.isClosed()) { LinkedBlockingQueue<WeakReference<TableResultIterator>> scannerQueue = @@ -3323,7 +3329,15 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement int renewed = 0; long start = System.currentTimeMillis(); while (numScanners > 0) { - WeakReference<TableResultIterator> ref = scannerQueue.poll(); + // It is guaranteed that this poll won't hang indefinitely because this is the + // only thread that removes items from the queue. Still adding a 1 ms timeout + // for sanity check. + WeakReference<TableResultIterator> ref = + scannerQueue.poll(1, TimeUnit.MILLISECONDS); + if (ref == null) { + throw new IllegalStateException( + "TableResulIterator ref found to be null. This is a bug. Some other thread removed items from the scanner queue."); + } TableResultIterator scanningItr = ref.get(); if (scanningItr != null) { RenewLeaseStatus status = scanningItr.renewLease(); http://git-wip-us.apache.org/repos/asf/phoenix/blob/6911770e/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java index 0b446b3..064007f 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java @@ -1450,25 +1450,26 @@ public class MetaDataClient { 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); + try (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()); + connection.rollback(); } - functionData.addAll(connection.getMutationState().toMutations().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(null).next().getSecond()); - connection.rollback(); + try (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(null).next().getSecond()); + connection.rollback(); + } MetaDataMutationResult result = connection.getQueryServices().createFunction(functionData, function, stmt.isTemporary()); MutationCode code = result.getMutationCode(); switch(code) { @@ -1880,7 +1881,6 @@ public class MetaDataClient { } } - PreparedStatement colUpsert = connection.prepareStatement(INSERT_COLUMN_CREATE_TABLE); Map<String, PName> familyNames = Maps.newLinkedHashMap(); boolean isPK = false; boolean rowTimeStampColumnAlreadyFound = false; @@ -2056,38 +2056,40 @@ public class MetaDataClient { } short nextKeySeq = 0; - for (int i = 0; i < columns.size(); i++) { - PColumn column = columns.get(i); - final int columnPosition = column.getPosition(); - // For client-side cache, we need to update the column - if (isViewColumnReferenced != null) { - if (viewColumnConstants != null && columnPosition < viewColumnConstants.length) { - columns.set(i, column = new DelegateColumn(column) { - @Override - public byte[] getViewConstant() { - return viewColumnConstants[columnPosition]; - } - @Override - public boolean isViewReferenced() { - return isViewColumnReferenced.get(columnPosition); - } - }); - } else { - columns.set(i, column = new DelegateColumn(column) { - @Override - public boolean isViewReferenced() { - return isViewColumnReferenced.get(columnPosition); - } - }); + + try (PreparedStatement colUpsert = connection.prepareStatement(INSERT_COLUMN_CREATE_TABLE)) { + for (int i = 0; i < columns.size(); i++) { + PColumn column = columns.get(i); + final int columnPosition = column.getPosition(); + // For client-side cache, we need to update the column + if (isViewColumnReferenced != null) { + if (viewColumnConstants != null && columnPosition < viewColumnConstants.length) { + columns.set(i, column = new DelegateColumn(column) { + @Override + public byte[] getViewConstant() { + return viewColumnConstants[columnPosition]; + } + @Override + public boolean isViewReferenced() { + return isViewColumnReferenced.get(columnPosition); + } + }); + } else { + columns.set(i, column = new DelegateColumn(column) { + @Override + public boolean isViewReferenced() { + return isViewColumnReferenced.get(columnPosition); + } + }); + } } + Short keySeq = SchemaUtil.isPKColumn(column) ? ++nextKeySeq : null; + addColumnMutation(schemaName, tableName, column, colUpsert, parentTableName, pkName, keySeq, saltBucketNum != null); } - Short keySeq = SchemaUtil.isPKColumn(column) ? ++nextKeySeq : null; - addColumnMutation(schemaName, tableName, column, colUpsert, parentTableName, pkName, keySeq, saltBucketNum != null); + tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond()); + connection.rollback(); } - tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond()); - connection.rollback(); - String dataTableName = parent == null || tableType == PTableType.VIEW ? null : parent.getTableName().getString(); PIndexState indexState = parent == null || tableType == PTableType.VIEW ? null : PIndexState.BUILDING; PreparedStatement tableUpsert = connection.prepareStatement(CREATE_TABLE); @@ -2572,12 +2574,13 @@ public class MetaDataClient { TABLE_NAME + "," + propertyName + ") VALUES (?, ?, ?, ?)"; - PreparedStatement tableBoolUpsert = connection.prepareStatement(updatePropertySql); - tableBoolUpsert.setString(1, tenantId); - tableBoolUpsert.setString(2, schemaName); - tableBoolUpsert.setString(3, tableName); - tableBoolUpsert.setBoolean(4, propertyValue); - tableBoolUpsert.execute(); + try (PreparedStatement tableBoolUpsert = connection.prepareStatement(updatePropertySql)) { + tableBoolUpsert.setString(1, tenantId); + tableBoolUpsert.setString(2, schemaName); + tableBoolUpsert.setString(3, tableName); + tableBoolUpsert.setBoolean(4, propertyValue); + tableBoolUpsert.execute(); + } } private void mutateLongProperty(String tenantId, String schemaName, String tableName, @@ -2588,12 +2591,13 @@ public class MetaDataClient { TABLE_NAME + "," + propertyName + ") VALUES (?, ?, ?, ?)"; - PreparedStatement tableBoolUpsert = connection.prepareStatement(updatePropertySql); - tableBoolUpsert.setString(1, tenantId); - tableBoolUpsert.setString(2, schemaName); - tableBoolUpsert.setString(3, tableName); - tableBoolUpsert.setLong(4, propertyValue); - tableBoolUpsert.execute(); + try (PreparedStatement tableBoolUpsert = connection.prepareStatement(updatePropertySql)) { + tableBoolUpsert.setString(1, tenantId); + tableBoolUpsert.setString(2, schemaName); + tableBoolUpsert.setString(3, tableName); + tableBoolUpsert.setLong(4, propertyValue); + tableBoolUpsert.execute(); + } } public MutationState addColumn(AddColumnStatement statement) throws SQLException { @@ -2743,77 +2747,76 @@ public class MetaDataClient { Long timeStamp = TransactionUtil.getTableTimestamp(connection, table.isTransactional() || nonTxToTx); int numPkColumnsAdded = 0; - PreparedStatement colUpsert = connection.prepareStatement(INSERT_COLUMN_ALTER_TABLE); - List<PColumn> columns = Lists.newArrayListWithExpectedSize(columnDefs.size()); Set<String> colFamiliesForPColumnsToBeAdded = new LinkedHashSet<>(); Set<String> families = new LinkedHashSet<>(); if (columnDefs.size() > 0 ) { - short nextKeySeq = SchemaUtil.getMaxKeySeq(table); - for( ColumnDef colDef : columnDefs) { - if (colDef != null && !colDef.isNull()) { - if(colDef.isPK()) { - throw new SQLExceptionInfo.Builder(SQLExceptionCode.NOT_NULLABLE_COLUMN_IN_ROW_KEY) - .setColumnName(colDef.getColumnDefName().getColumnName()).build().buildException(); - } else { - throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ADD_NOT_NULLABLE_COLUMN) + try (PreparedStatement colUpsert = connection.prepareStatement(INSERT_COLUMN_ALTER_TABLE)) { + short nextKeySeq = SchemaUtil.getMaxKeySeq(table); + for( ColumnDef colDef : columnDefs) { + if (colDef != null && !colDef.isNull()) { + if(colDef.isPK()) { + throw new SQLExceptionInfo.Builder(SQLExceptionCode.NOT_NULLABLE_COLUMN_IN_ROW_KEY) + .setColumnName(colDef.getColumnDefName().getColumnName()).build().buildException(); + } else { + throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_ADD_NOT_NULLABLE_COLUMN) + .setColumnName(colDef.getColumnDefName().getColumnName()).build().buildException(); + } + } + if (colDef != null && colDef.isPK() && table.getType() == VIEW && table.getViewType() != MAPPED) { + throwIfLastPKOfParentIsFixedLength(getParentOfView(table), schemaName, tableName, colDef); + } + if (colDef != null && colDef.isRowTimestamp()) { + throw new SQLExceptionInfo.Builder(SQLExceptionCode.ROWTIMESTAMP_CREATE_ONLY) .setColumnName(colDef.getColumnDefName().getColumnName()).build().buildException(); } - } - if (colDef != null && colDef.isPK() && table.getType() == VIEW && table.getViewType() != MAPPED) { - throwIfLastPKOfParentIsFixedLength(getParentOfView(table), schemaName, tableName, colDef); - } - if (colDef != null && colDef.isRowTimestamp()) { - throw new SQLExceptionInfo.Builder(SQLExceptionCode.ROWTIMESTAMP_CREATE_ONLY) - .setColumnName(colDef.getColumnDefName().getColumnName()).build().buildException(); - } - PColumn column = newColumn(position++, colDef, PrimaryKeyConstraint.EMPTY, table.getDefaultFamilyName() == null ? null : table.getDefaultFamilyName().getString(), true); - columns.add(column); - String pkName = null; - Short keySeq = null; - - // TODO: support setting properties on other families? - if (column.getFamilyName() == null) { - ++numPkColumnsAdded; - pkName = table.getPKName() == null ? null : table.getPKName().getString(); - keySeq = ++nextKeySeq; - } else { - families.add(column.getFamilyName().getString()); - } - colFamiliesForPColumnsToBeAdded.add(column.getFamilyName() == null ? null : column.getFamilyName().getString()); - addColumnMutation(schemaName, tableName, column, colUpsert, null, pkName, keySeq, table.getBucketNum() != null); - } - - // Add any new PK columns to end of index PK - if (numPkColumnsAdded>0) { - // create PK column list that includes the newly created columns - List<PColumn> pkColumns = Lists.newArrayListWithExpectedSize(table.getPKColumns().size()+numPkColumnsAdded); - pkColumns.addAll(table.getPKColumns()); - for (int i=0; i<columnDefs.size(); ++i) { - if (columnDefs.get(i).isPK()) { - pkColumns.add(columns.get(i)); + PColumn column = newColumn(position++, colDef, PrimaryKeyConstraint.EMPTY, table.getDefaultFamilyName() == null ? null : table.getDefaultFamilyName().getString(), true); + columns.add(column); + String pkName = null; + Short keySeq = null; + + // TODO: support setting properties on other families? + if (column.getFamilyName() == null) { + ++numPkColumnsAdded; + pkName = table.getPKName() == null ? null : table.getPKName().getString(); + keySeq = ++nextKeySeq; + } else { + families.add(column.getFamilyName().getString()); } + colFamiliesForPColumnsToBeAdded.add(column.getFamilyName() == null ? null : column.getFamilyName().getString()); + addColumnMutation(schemaName, tableName, column, colUpsert, null, pkName, keySeq, table.getBucketNum() != null); } - int pkSlotPosition = table.getPKColumns().size()-1; - for (PTable index : table.getIndexes()) { - short nextIndexKeySeq = SchemaUtil.getMaxKeySeq(index); - int indexPosition = index.getColumns().size(); + + // Add any new PK columns to end of index PK + if (numPkColumnsAdded>0) { + // create PK column list that includes the newly created columns + List<PColumn> pkColumns = Lists.newArrayListWithExpectedSize(table.getPKColumns().size()+numPkColumnsAdded); + pkColumns.addAll(table.getPKColumns()); for (int i=0; i<columnDefs.size(); ++i) { - ColumnDef colDef = columnDefs.get(i); - if (colDef.isPK()) { - PDataType indexColDataType = IndexUtil.getIndexColumnDataType(colDef.isNull(), colDef.getDataType()); - ColumnName indexColName = ColumnName.caseSensitiveColumnName(IndexUtil.getIndexColumnName(null, colDef.getColumnDefName().getColumnName())); - Expression expression = new RowKeyColumnExpression(columns.get(i), new RowKeyValueAccessor(pkColumns, ++pkSlotPosition)); - ColumnDef indexColDef = FACTORY.columnDef(indexColName, indexColDataType.getSqlTypeName(), colDef.isNull(), colDef.getMaxLength(), colDef.getScale(), true, colDef.getSortOrder(), expression.toString(), colDef.isRowTimestamp()); - PColumn indexColumn = newColumn(indexPosition++, indexColDef, PrimaryKeyConstraint.EMPTY, null, true); - addColumnMutation(schemaName, index.getTableName().getString(), indexColumn, colUpsert, index.getParentTableName().getString(), index.getPKName() == null ? null : index.getPKName().getString(), ++nextIndexKeySeq, index.getBucketNum() != null); + if (columnDefs.get(i).isPK()) { + pkColumns.add(columns.get(i)); + } + } + int pkSlotPosition = table.getPKColumns().size()-1; + for (PTable index : table.getIndexes()) { + short nextIndexKeySeq = SchemaUtil.getMaxKeySeq(index); + int indexPosition = index.getColumns().size(); + for (int i=0; i<columnDefs.size(); ++i) { + ColumnDef colDef = columnDefs.get(i); + if (colDef.isPK()) { + PDataType indexColDataType = IndexUtil.getIndexColumnDataType(colDef.isNull(), colDef.getDataType()); + ColumnName indexColName = ColumnName.caseSensitiveColumnName(IndexUtil.getIndexColumnName(null, colDef.getColumnDefName().getColumnName())); + Expression expression = new RowKeyColumnExpression(columns.get(i), new RowKeyValueAccessor(pkColumns, ++pkSlotPosition)); + ColumnDef indexColDef = FACTORY.columnDef(indexColName, indexColDataType.getSqlTypeName(), colDef.isNull(), colDef.getMaxLength(), colDef.getScale(), true, colDef.getSortOrder(), expression.toString(), colDef.isRowTimestamp()); + PColumn indexColumn = newColumn(indexPosition++, indexColDef, PrimaryKeyConstraint.EMPTY, null, true); + addColumnMutation(schemaName, index.getTableName().getString(), indexColumn, colUpsert, index.getParentTableName().getString(), index.getPKName() == null ? null : index.getPKName().getString(), ++nextIndexKeySeq, index.getBucketNum() != null); + } } } } + columnMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond()); + connection.rollback(); } - - columnMetaData.addAll(connection.getMutationState().toMutations(timeStamp).next().getSecond()); - connection.rollback(); } else { // Check that HBase configured properly for mutable secondary indexing // if we're changing from an immutable table to a mutable table and we http://git-wip-us.apache.org/repos/asf/phoenix/blob/6911770e/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index 7191b1e..e635ae7 100644 --- a/pom.xml +++ b/pom.xml @@ -406,6 +406,7 @@ <argLine>-enableassertions -Xmx2250m -XX:MaxPermSize=128m -Djava.security.egd=file:/dev/./urandom "-Djava.library.path=${hadoop.library.path}${path.separator}${java.library.path}"</argLine> <redirectTestOutputToFile>${test.output.tofile}</redirectTestOutputToFile> + <shutdown>kill</shutdown> </configuration> </plugin> <!-- All projects create a test jar -->
