You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2020/03/04 12:08:01 UTC
[incubator-iotdb] branch master updated: add filed nullable check
(#881)
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9fc13b6 add filed nullable check (#881)
9fc13b6 is described below
commit 9fc13b6bc46128b5f64e74f3a59cababd456ea17
Author: Dawei Liu <at...@163.com>
AuthorDate: Wed Mar 4 20:07:51 2020 +0800
add filed nullable check (#881)
* add filed nullable check
---
.../iotdb/hadoop/tsfile/TSFRecordReader.java | 137 +++++++++++----------
.../watermark/GroupedLSBWatermarkEncoder.java | 2 +-
.../apache/iotdb/db/utils/QueryDataSetUtils.java | 24 ++--
3 files changed, 87 insertions(+), 76 deletions(-)
diff --git a/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TSFRecordReader.java b/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TSFRecordReader.java
index 146b80b..15b6e8e 100644
--- a/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TSFRecordReader.java
+++ b/hadoop/src/main/java/org/apache/iotdb/hadoop/tsfile/TSFRecordReader.java
@@ -57,15 +57,15 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
*/
private List<QueryDataSet> dataSetList = new ArrayList<>();
/**
- * List for name of devices. The order corresponds to the order of dataSetList.
- * Means that deviceIdList[i] is the name of device for dataSetList[i].
+ * List for name of devices. The order corresponds to the order of dataSetList. Means that
+ * deviceIdList[i] is the name of device for dataSetList[i].
*/
private List<String> deviceIdList = new ArrayList<>();
private List<Field> fields = null;
/**
* The index of QueryDataSet that is currently processed
*/
- private int currentIndex = 0;
+ private int currentIndex = 0;
private long timestamp = 0;
private boolean isReadDeviceId = false;
private boolean isReadTime = false;
@@ -77,67 +77,71 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
public void initialize(InputSplit split, TaskAttemptContext context)
throws IOException {
if (split instanceof TSFInputSplit) {
- initialize((TSFInputSplit) split, context.getConfiguration(), this, dataSetList, deviceIdList);
- }
- else {
+ initialize((TSFInputSplit) split, context.getConfiguration(), this, dataSetList,
+ deviceIdList);
+ } else {
logger.error("The InputSplit class is not {}, the class is {}", TSFInputSplit.class.getName(),
- split.getClass().getName());
+ split.getClass().getName());
throw new InternalError(String.format("The InputSplit class is not %s, the class is %s",
- TSFInputSplit.class.getName(), split.getClass().getName()));
+ TSFInputSplit.class.getName(), split.getClass().getName()));
}
}
- public static void initialize(TSFInputSplit split, Configuration configuration, IReaderSet readerSet, List<QueryDataSet> dataSetList, List<String> deviceIdList) throws IOException {
- org.apache.hadoop.fs.Path path = split.getPath();
- List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList = split.getChunkGroupInfoList();
- TsFileSequenceReader reader = new TsFileSequenceReader(new HDFSInput(path, configuration));
- readerSet.setReader(reader);
- // Get the read columns and filter information
-
- List<String> deviceIds = TSFInputFormat.getReadDeviceIds(configuration);
- if (deviceIds == null) {
- deviceIds = initDeviceIdList(chunkGroupInfoList);
- }
- List<String> measurementIds = TSFInputFormat.getReadMeasurementIds(configuration);
- if (measurementIds == null) {
- measurementIds = initSensorIdList(chunkGroupInfoList);
- }
- readerSet.setMeasurementIds(measurementIds);
- logger.info("deviceIds:" + deviceIds);
- logger.info("Sensors:" + measurementIds);
+ public static void initialize(TSFInputSplit split, Configuration configuration,
+ IReaderSet readerSet, List<QueryDataSet> dataSetList, List<String> deviceIdList)
+ throws IOException {
+ org.apache.hadoop.fs.Path path = split.getPath();
+ List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList = split.getChunkGroupInfoList();
+ TsFileSequenceReader reader = new TsFileSequenceReader(new HDFSInput(path, configuration));
+ readerSet.setReader(reader);
+ // Get the read columns and filter information
+ List<String> deviceIds = TSFInputFormat.getReadDeviceIds(configuration);
+ if (deviceIds == null) {
+ deviceIds = initDeviceIdList(chunkGroupInfoList);
+ }
+ List<String> measurementIds = TSFInputFormat.getReadMeasurementIds(configuration);
+ if (measurementIds == null) {
+ measurementIds = initSensorIdList(chunkGroupInfoList);
+ }
+ readerSet.setMeasurementIds(measurementIds);
+ logger.info("deviceIds:" + deviceIds);
+ logger.info("Sensors:" + measurementIds);
- readerSet.setReadDeviceId(TSFInputFormat.getReadDeviceId(configuration));
- readerSet.setReadTime(TSFInputFormat.getReadTime(configuration));
+ readerSet.setReadDeviceId(TSFInputFormat.getReadDeviceId(configuration));
+ readerSet.setReadTime(TSFInputFormat.getReadTime(configuration));
- ReadOnlyTsFile queryEngine = new ReadOnlyTsFile(reader);
- for (TSFInputSplit.ChunkGroupInfo chunkGroupInfo : chunkGroupInfoList) {
- String deviceId = chunkGroupInfo.getDeviceId();
- if (deviceIds.contains(deviceId)) {
- List<Path> paths = measurementIds.stream()
- .map(measurementId -> new Path(deviceId + TsFileConstant.PATH_SEPARATOR + measurementId))
- .collect(toList());
- QueryExpression queryExpression = QueryExpression.create(paths, null);
- QueryDataSet dataSet = queryEngine.query(queryExpression,
- chunkGroupInfo.getStartOffset(), chunkGroupInfo.getEndOffset());
- dataSetList.add(dataSet);
- deviceIdList.add(deviceId);
- }
+ ReadOnlyTsFile queryEngine = new ReadOnlyTsFile(reader);
+ for (TSFInputSplit.ChunkGroupInfo chunkGroupInfo : chunkGroupInfoList) {
+ String deviceId = chunkGroupInfo.getDeviceId();
+ if (deviceIds.contains(deviceId)) {
+ List<Path> paths = measurementIds.stream()
+ .map(
+ measurementId -> new Path(deviceId + TsFileConstant.PATH_SEPARATOR + measurementId))
+ .collect(toList());
+ QueryExpression queryExpression = QueryExpression.create(paths, null);
+ QueryDataSet dataSet = queryEngine.query(queryExpression,
+ chunkGroupInfo.getStartOffset(), chunkGroupInfo.getEndOffset());
+ dataSetList.add(dataSet);
+ deviceIdList.add(deviceId);
}
+ }
}
- private static List<String> initDeviceIdList(List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList) {
+ private static List<String> initDeviceIdList(
+ List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList) {
return chunkGroupInfoList.stream()
- .map(TSFInputSplit.ChunkGroupInfo::getDeviceId)
- .distinct()
- .collect(toList());
+ .map(TSFInputSplit.ChunkGroupInfo::getDeviceId)
+ .distinct()
+ .collect(toList());
}
- private static List<String> initSensorIdList(List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList) {
+ private static List<String> initSensorIdList(
+ List<TSFInputSplit.ChunkGroupInfo> chunkGroupInfoList) {
return chunkGroupInfoList.stream()
- .flatMap(chunkGroupMetaData -> Arrays.stream(chunkGroupMetaData.getMeasurementIds()))
- .distinct()
- .collect(toList());
+ .flatMap(chunkGroupMetaData -> Arrays.stream(chunkGroupMetaData.getMeasurementIds()))
+ .distinct()
+ .collect(toList());
}
@Override
@@ -145,8 +149,7 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
while (currentIndex < dataSetList.size()) {
if (!dataSetList.get(currentIndex).hasNext()) {
currentIndex++;
- }
- else {
+ } else {
RowRecord rowRecord = dataSetList.get(currentIndex).next();
fields = rowRecord.getFields();
timestamp = rowRecord.getTimestamp();
@@ -163,13 +166,14 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
@Override
public MapWritable getCurrentValue() throws InterruptedException {
- return getCurrentValue(deviceIdList, currentIndex, timestamp, isReadTime, isReadDeviceId, fields, measurementIds);
+ return getCurrentValue(deviceIdList, currentIndex, timestamp, isReadTime, isReadDeviceId,
+ fields, measurementIds);
}
public static MapWritable getCurrentValue(List<String> deviceIdList, int currentIndex,
- long timestamp, boolean isReadTime,
- boolean isReadDeviceId, List<Field> fields,
- List<String> measurementIds) throws InterruptedException {
+ long timestamp, boolean isReadTime,
+ boolean isReadDeviceId, List<Field> fields,
+ List<String> measurementIds) throws InterruptedException {
MapWritable mapWritable = new MapWritable();
Text deviceIdText = new Text(deviceIdList.get(currentIndex));
LongWritable time = new LongWritable(timestamp);
@@ -188,13 +192,15 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
/**
* Read from current fields value
+ *
* @param mapWritable where to write
* @throws InterruptedException
*/
- public static void readFieldsValue(MapWritable mapWritable, List<Field> fields, List<String> measurementIds) throws InterruptedException {
+ public static void readFieldsValue(MapWritable mapWritable, List<Field> fields,
+ List<String> measurementIds) throws InterruptedException {
int index = 0;
for (Field field : fields) {
- if (field.getDataType() == null) {
+ if (field == null || field.getDataType() == null) {
logger.info("Current value is null");
mapWritable.put(new Text(measurementIds.get(index)), NullWritable.get());
} else {
@@ -203,24 +209,29 @@ public class TSFRecordReader extends RecordReader<NullWritable, MapWritable> imp
mapWritable.put(new Text(measurementIds.get(index)), new IntWritable(field.getIntV()));
break;
case INT64:
- mapWritable.put(new Text(measurementIds.get(index)), new LongWritable(field.getLongV()));
+ mapWritable
+ .put(new Text(measurementIds.get(index)), new LongWritable(field.getLongV()));
break;
case FLOAT:
- mapWritable.put(new Text(measurementIds.get(index)), new FloatWritable(field.getFloatV()));
+ mapWritable
+ .put(new Text(measurementIds.get(index)), new FloatWritable(field.getFloatV()));
break;
case DOUBLE:
- mapWritable.put(new Text(measurementIds.get(index)), new DoubleWritable(field.getDoubleV()));
+ mapWritable
+ .put(new Text(measurementIds.get(index)), new DoubleWritable(field.getDoubleV()));
break;
case BOOLEAN:
- mapWritable.put(new Text(measurementIds.get(index)), new BooleanWritable(field.getBoolV()));
+ mapWritable
+ .put(new Text(measurementIds.get(index)), new BooleanWritable(field.getBoolV()));
break;
case TEXT:
- mapWritable.put(new Text(measurementIds.get(index)), new Text(field.getBinaryV().getStringValue()));
+ mapWritable.put(new Text(measurementIds.get(index)),
+ new Text(field.getBinaryV().getStringValue()));
break;
default:
logger.error("The data type is not support {}", field.getDataType());
throw new InterruptedException(
- String.format("The data type %s is not support ", field.getDataType()));
+ String.format("The data type %s is not support ", field.getDataType()));
}
}
index++;
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/watermark/GroupedLSBWatermarkEncoder.java b/server/src/main/java/org/apache/iotdb/db/tools/watermark/GroupedLSBWatermarkEncoder.java
index 842290e..4c6d3b6 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/watermark/GroupedLSBWatermarkEncoder.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/watermark/GroupedLSBWatermarkEncoder.java
@@ -113,7 +113,7 @@ public class GroupedLSBWatermarkEncoder implements WatermarkEncoder {
}
List<Field> fields = record.getFields();
for (Field field : fields) {
- if (field.getDataType() == null) {
+ if (field == null || field.getDataType() == null) {
continue;
}
TSDataType dataType = field.getDataType();
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
index 4209ba3..c5a9114 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryDataSetUtils.java
@@ -49,7 +49,7 @@ public class QueryDataSetUtils {
* convert query data set by fetch size.
*
* @param queryDataSet -query dataset
- * @param fetchSize -fetch size
+ * @param fetchSize -fetch size
* @return -convert query dataset
*/
public static TSQueryDataSet convertQueryDataSetByFetchSize(QueryDataSet queryDataSet,
@@ -86,11 +86,11 @@ public class QueryDataSetUtils {
List<Field> fields = rowRecord.getFields();
for (int k = 0; k < fields.size(); k++) {
Field field = fields.get(k);
- DataOutputStream dataOutputStream = dataOutputStreams[2*k + 1]; // DO NOT FORGET +1
- if (field.getDataType() == null) {
- bitmap[k] = (bitmap[k] << 1);
+ DataOutputStream dataOutputStream = dataOutputStreams[2 * k + 1]; // DO NOT FORGET +1
+ if (field == null || field.getDataType() == null) {
+ bitmap[k] = (bitmap[k] << 1);
} else {
- bitmap[k] = (bitmap[k] << 1) | flag;
+ bitmap[k] = (bitmap[k] << 1) | flag;
TSDataType type = field.getDataType();
switch (type) {
case INT32:
@@ -127,7 +127,7 @@ public class QueryDataSetUtils {
rowCount++;
if (rowCount % 8 == 0) {
for (int j = 0; j < bitmap.length; j++) {
- DataOutputStream dataBitmapOutputStream = dataOutputStreams[2*(j+1)];
+ DataOutputStream dataBitmapOutputStream = dataOutputStreams[2 * (j + 1)];
dataBitmapOutputStream.writeByte(bitmap[j]);
// we should clear the bitmap every 8 row record
bitmap[j] = 0;
@@ -142,8 +142,8 @@ public class QueryDataSetUtils {
int remaining = rowCount % 8;
if (remaining != 0) {
for (int j = 0; j < bitmap.length; j++) {
- DataOutputStream dataBitmapOutputStream = dataOutputStreams[2*(j+1)];
- dataBitmapOutputStream.writeByte(bitmap[j] << (8-remaining));
+ DataOutputStream dataBitmapOutputStream = dataOutputStreams[2 * (j + 1)];
+ dataBitmapOutputStream.writeByte(bitmap[j] << (8 - remaining));
}
}
@@ -160,13 +160,13 @@ public class QueryDataSetUtils {
List<ByteBuffer> bitmapList = new LinkedList<>();
List<ByteBuffer> valueList = new LinkedList<>();
for (int i = 1; i < byteArrayOutputStreams.length; i += 2) {
- ByteBuffer valueBuffer = ByteBuffer.allocate(valueOccupation[(i-1)/2]);
+ ByteBuffer valueBuffer = ByteBuffer.allocate(valueOccupation[(i - 1) / 2]);
valueBuffer.put(byteArrayOutputStreams[i].toByteArray());
valueBuffer.flip();
valueList.add(valueBuffer);
ByteBuffer bitmapBuffer = ByteBuffer.allocate(bitmapOccupation);
- bitmapBuffer.put(byteArrayOutputStreams[i+1].toByteArray());
+ bitmapBuffer.put(byteArrayOutputStreams[i + 1].toByteArray());
bitmapBuffer.flip();
bitmapList.add(bitmapBuffer);
}
@@ -194,9 +194,9 @@ public class QueryDataSetUtils {
}
/**
- * @param buffer data values
+ * @param buffer data values
* @param columns column number
- * @param size value count in each column
+ * @param size value count in each column
*/
public static Object[] readValuesFromBuffer(ByteBuffer buffer, TSDataType[] types,
int columns, int size) {