natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-1.2' into trunk

This commit is contained in:
Jonathan Ellis 2013-07-15 17:44:46 -05:00
parent 3f87907697 b2daab711c
commit e4ebd5ff59
1 changed files with 23 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
src/java/org/apache/cassandra/hadoop/cql3/CqlPagingRecordReader.java
View File

@ -186,7 +186,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
{
if (!rowIterator.hasNext())
{
logger.debug("Finished scanning " + rowIterator.totalRead + " rows (estimate was: " + totalRowCount + ")");
logger.debug("Finished scanning {} rows (estimate was: {})", rowIterator.totalRead, totalRowCount);
return false;
}
@ -295,18 +295,18 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
// no more data
if (index == -1 || emptyPartitionKeyValues())
{
logger.debug("no more data.");
logger.debug("no more data");
return endOfData();
}
index = setTailNull(clusterColumns);
logger.debug("set tail to null, index: " + index);
logger.debug("set tail to null, index: {}", index);
executeQuery();
pageRows = 0;
if (rows == null || !rows.hasNext() && index < 0)
{
logger.debug("no more data.");
logger.debug("no more data");
return endOfData();
}
}
@ -318,7 +318,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
for (Column column : row.columns)
{
String columnName = stringValue(ByteBuffer.wrap(column.getName()));
logger.debug("column: " + columnName);
logger.debug("column: {}", columnName);
if (i < partitionBoundColumns.size() + clusterColumns.size())
keyColumns.put(stringValue(column.name), column.value);
@ -370,7 +370,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
rowKey = rowKey + column.validator.getString(ByteBufferUtil.clone(iter.next())) + ":";
}
logger.debug("previous RowKey: " + previousRowKey + ", new row key: " + rowKey);
logger.debug("previous RowKey: {}, new row key: {}", previousRowKey, rowKey);
if (previousRowKey == null)
{
this.previousRowKey = rowKey;
@ -400,7 +400,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
{
int index = previousIndex > 0 ? previousIndex : 0;
BoundColumn column = values.get(index);
logger.debug("set key " + column.name + " value to null");
logger.debug("set key {} value to null", column.name);
column.value = null;
return previousIndex - 1;
}
@ -409,7 +409,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
}
BoundColumn column = values.get(previousIndex);
logger.debug("set key " + column.name + " value to null");
logger.debug("set key {} value to null", column.name);
column.value = null;
return previousIndex - 1;
}
@ -434,13 +434,10 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
: quote(partitionKey) + "," + quote(clusterKey) + "," + columns;
}
String whereStr = userDefinedWhereClauses == null ? "" : " AND " + userDefinedWhereClauses;
return Pair.create(clause.left,
"SELECT " + columns
+ " FROM " + quote(cfName)
+ clause.right
+ (userDefinedWhereClauses == null ? "" : " AND " + userDefinedWhereClauses)
+ " LIMIT " + pageRowSize
+ " ALLOW FILTERING");
String.format("SELECT %s FROM %s%s%s LIMIT %d ALLOW FILTERING",
columns, quote(cfName), clause.right, whereStr, pageRowSize));
}
@ -475,28 +472,28 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
partitionKeyMarkers = partitionKeyMarkers();
// initial query token(k) >= start_token and token(k) <= end_token
if (emptyPartitionKeyValues())
return Pair.create(0, " WHERE token(" + partitionKeyString + ") > ? AND token(" + partitionKeyString + ") <= ?");
return Pair.create(0, String.format(" WHERE token(%s) > ? AND token(%s) <= ?", partitionKeyString, partitionKeyString));
// query token(k) > token(pre_partition_key) and token(k) <= end_token
if (clusterColumns.size() == 0 || clusterColumns.get(0).value == null)
return Pair.create(1,
" WHERE token(" + partitionKeyString + ") > token(" + partitionKeyMarkers + ") "
+ " AND token(" + partitionKeyString + ") <= ?");
String.format(" WHERE token(%s) > token(%s) AND token(%s) <= ?",
partitionKeyString, partitionKeyMarkers, partitionKeyString));
// query token(k) = token(pre_partition_key) and m = pre_cluster_key_m and n > pre_cluster_key_n
Pair<Integer, String> clause = whereClause(clusterColumns, 0);
return Pair.create(clause.left,
" WHERE token(" + partitionKeyString + ") = token(" + partitionKeyMarkers + ") " + clause.right);
String.format(" WHERE token(%s) = token(%s) %s", partitionKeyString, partitionKeyMarkers, clause.right));
}
/** recursively serialize the where clause */
private Pair<Integer, String> whereClause(List<BoundColumn> column, int position)
{
if (position == column.size() - 1 || column.get(position + 1).value == null)
return Pair.create(position + 2, " AND " + quote(column.get(position).name) + " > ? ");
return Pair.create(position + 2, String.format(" AND %s > ? ", quote(column.get(position).name)));
Pair<Integer, String> clause = whereClause(column, position + 1);
return Pair.create(clause.left, " AND " + quote(column.get(position).name) + " = ? " + clause.right);
return Pair.create(clause.left, String.format(" AND %s = ? %s", quote(column.get(position).name), clause.right));
}
/** check whether all key values are null */
@ -586,7 +583,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
Pair<Integer, String> query = null;
query = composeQuery(columns);
logger.debug("type:" + query.left + ", query: " + query.right);
logger.debug("type: {}, query: {}", query.left, query.right);
CqlPreparedResult cqlPreparedResult = client.prepare_cql3_query(ByteBufferUtil.bytes(query.right), Compression.NONE);
preparedQueryIds.put(query.left, cqlPreparedResult.itemId);
return cqlPreparedResult.itemId;
@ -601,7 +598,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
private void executeQuery()
{
Pair<Integer, List<ByteBuffer>> bindValues = preparedQueryBindValues();
logger.debug("query type: " + bindValues.left);
logger.debug("query type: {}", bindValues.left);
// check whether it reach end of range for type 1 query CASSANDRA-5573
if (bindValues.left == 1 && reachEndRange())
@ -668,14 +665,14 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
CqlRow cqlRow = result.rows.get(0);
String keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(0).getValue()));
logger.debug("partition keys: " + keyString);
logger.debug("partition keys: {}", keyString);
List<String> keys = FBUtilities.fromJsonList(keyString);
for (String key : keys)
partitionBoundColumns.add(new BoundColumn(key));
keyString = ByteBufferUtil.string(ByteBuffer.wrap(cqlRow.columns.get(1).getValue()));
logger.debug("cluster columns: " + keyString);
logger.debug("cluster columns: {}", keyString);
keys = FBUtilities.fromJsonList(keyString);
for (String key : keys)
@ -683,7 +680,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
Column rawKeyValidator = cqlRow.columns.get(2);
String validator = ByteBufferUtil.string(ByteBuffer.wrap(rawKeyValidator.getValue()));
logger.debug("row key validator: " + validator);
logger.debug("row key validator: {}", validator);
keyValidator = parseType(validator);
if (keyValidator instanceof CompositeType)
@ -718,7 +715,7 @@ public class CqlPagingRecordReader extends RecordReader<Map<String, ByteBuffer>,
String endToken = split.getEndToken();
String currentToken = partitioner.getToken(rowKey).toString();
logger.debug("End token: " + endToken + ", current token: " + currentToken);
logger.debug("End token: {}, current token: {}", endToken, currentToken);
return endToken.equals(currentToken);
}