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 68264c3 fix IOTDB-235 (#419)
68264c3 is described below
commit 68264c346e4c3a0f7936cea07eaeaa2c76e4d5a3
Author: SilverNarcissus <[email protected]>
AuthorDate: Thu Sep 26 10:02:39 2019 +0800
fix IOTDB-235 (#419)
---
.../db/engine/storagegroup/TsFileProcessor.java | 2 +
.../iotdb/db/qp/physical/crud/BatchInsertPlan.java | 66 ++++++++++++----------
.../iotdb/db/writelog/recover/LogReplayer.java | 41 ++++++++++++--
3 files changed, 73 insertions(+), 36 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 82786ee..65a141f 100644
---
a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++
b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.engine.storagegroup;
import java.io.File;
import java.io.IOException;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedDeque;
@@ -170,6 +171,7 @@ public class TsFileProcessor {
if (IoTDBDescriptor.getInstance().getConfig().isEnableWal()) {
try {
+ batchInsertPlan.setIndex(new HashSet<>(indexes));
getLogNode().write(batchInsertPlan);
} catch (IOException e) {
logger.error("write WAL failed", e);
diff --git
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
index ce907ed..7433565 100644
---
a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
+++
b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/BatchInsertPlan.java
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.qp.physical.crud;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
@@ -41,18 +42,15 @@ public class BatchInsertPlan extends PhysicalPlan {
private Object[] columns;
private ByteBuffer valueBuffer;
-
+ private Set<Integer> index;
private int rowCount = 0;
-
// cached values
private Long maxTime = null;
private Long minTime = null;
private List<Path> paths;
-
public BatchInsertPlan() {
super(false, OperatorType.BATCHINSERT);
}
-
public BatchInsertPlan(String deviceId, List<String> measurements) {
super(false, OperatorType.BATCHINSERT);
this.deviceId = deviceId;
@@ -66,6 +64,13 @@ public class BatchInsertPlan extends PhysicalPlan {
setDataTypes(dataTypes);
}
+ public Set<Integer> getIndex() {
+ return index;
+ }
+
+ public void setIndex(Set<Integer> index) {
+ this.index = index;
+ }
@Override
public List<Path> getPaths() {
@@ -92,15 +97,15 @@ public class BatchInsertPlan extends PhysicalPlan {
putString(buffer, m);
}
- for (TSDataType dataType: dataTypes) {
+ for (TSDataType dataType : dataTypes) {
buffer.putShort(dataType.serialize());
}
- buffer.putInt(times.length);
+ buffer.putInt(index.size());
if (timeBuffer == null) {
- for (long time: times) {
- buffer.putLong(time);
+ for(int loc : index){
+ buffer.putLong(times[loc]);
}
} else {
buffer.put(timeBuffer.array());
@@ -113,39 +118,39 @@ public class BatchInsertPlan extends PhysicalPlan {
switch (dataType) {
case INT32:
int[] intValues = (int[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.putInt(intValues[index]);
+ for(int loc : index){
+ buffer.putInt(intValues[loc]);
}
break;
case INT64:
long[] longValues = (long[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.putLong(longValues[index]);
+ for(int loc : index){
+ buffer.putLong(longValues[loc]);
}
break;
case FLOAT:
float[] floatValues = (float[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.putFloat(floatValues[index]);
+ for(int loc : index){
+ buffer.putFloat(floatValues[loc]);
}
break;
case DOUBLE:
double[] doubleValues = (double[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.putDouble(doubleValues[index]);
+ for(int loc : index){
+ buffer.putDouble(doubleValues[loc]);
}
break;
case BOOLEAN:
boolean[] boolValues = (boolean[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.put(BytesUtils.boolToByte(boolValues[index]));
+ for(int loc : index){
+ buffer.put(BytesUtils.boolToByte(boolValues[loc]));
}
break;
case TEXT:
Binary[] binaryValues = (Binary[]) columns[i];
- for (int index = 0; index < rowCount; index++) {
- buffer.putInt(binaryValues[index].getLength());
- buffer.put(binaryValues[index].getValues());
+ for(int loc : index){
+ buffer.putInt(binaryValues[loc].getLength());
+ buffer.put(binaryValues[loc].getValues());
}
break;
default:
@@ -185,10 +190,11 @@ public class BatchInsertPlan extends PhysicalPlan {
}
int rows = buffer.getInt();
+ rowCount = rows;
this.times = new long[rows];
- QueryDataSetUtils.readTimesFromBuffer(buffer, rows);
+ times = QueryDataSetUtils.readTimesFromBuffer(buffer, rows);
- QueryDataSetUtils.readValuesFromBuffer(buffer, dataTypes, measurementSize,
rows);
+ columns = QueryDataSetUtils.readValuesFromBuffer(buffer, dataTypes,
measurementSize, rows);
}
@@ -228,12 +234,16 @@ public class BatchInsertPlan extends PhysicalPlan {
return columns;
}
+ public void setColumns(Object[] columns) {
+ this.columns = columns;
+ }
+
public long getMinTime() {
if (minTime != null) {
return minTime;
}
minTime = Long.MAX_VALUE;
- for (Long time: times) {
+ for (Long time : times) {
if (time < minTime) {
minTime = time;
}
@@ -246,7 +256,7 @@ public class BatchInsertPlan extends PhysicalPlan {
return maxTime;
}
long maxTime = Long.MIN_VALUE;
- for (Long time: times) {
+ for (Long time : times) {
if (time > maxTime) {
maxTime = time;
}
@@ -269,8 +279,4 @@ public class BatchInsertPlan extends PhysicalPlan {
public void setRowCount(int size) {
this.rowCount = size;
}
-
- public void setColumns(Object[] columns) {
- this.columns = columns;
- }
}
diff --git
a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
index fd39356..4fbb3f7 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
@@ -7,7 +7,7 @@
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
- * http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
@@ -20,10 +20,10 @@
package org.apache.iotdb.db.writelog.recover;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-
import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.ModificationFile;
@@ -32,6 +32,7 @@ import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.exception.qp.QueryProcessorException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.BatchInsertPlan;
import org.apache.iotdb.db.qp.physical.crud.DeletePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
import org.apache.iotdb.db.qp.physical.crud.UpdatePlan;
@@ -48,7 +49,6 @@ import org.apache.iotdb.tsfile.write.schema.Schema;
* the WALs from the logNode and redoes them into a given MemTable and
ModificationFile.
*/
public class LogReplayer {
-
private String logNodePrefix;
private String insertFilePath;
private ModificationFile modFile;
@@ -98,10 +98,14 @@ public class LogReplayer {
replayDelete((DeletePlan) plan);
} else if (plan instanceof UpdatePlan) {
replayUpdate((UpdatePlan) plan);
+ } else if (plan instanceof BatchInsertPlan) {
+ replayBatchInsert((BatchInsertPlan) plan);
}
}
- } catch (IOException | QueryProcessorException e) {
+ } catch (IOException e) {
throw new ProcessorException("Cannot replay logs", e);
+ } catch (QueryProcessorException e) {
+ throw new ProcessorException("Cannot replay logs for query processor
exception", e);
} finally {
logReader.close();
}
@@ -113,15 +117,40 @@ public class LogReplayer {
List<Path> paths = deletePlan.getPaths();
for (Path path : paths) {
recoverMemTable.delete(path.getDevice(), path.getMeasurement(),
deletePlan.getDeleteTime());
- modFile.write(new Deletion(path,
versionController.nextVersion(),deletePlan.getDeleteTime()));
+ modFile
+ .write(new Deletion(path, versionController.nextVersion(),
deletePlan.getDeleteTime()));
+ }
+ }
+
+ private void replayBatchInsert(BatchInsertPlan batchInsertPlan) throws
QueryProcessorException {
+ if (currentTsFileResource != null) {
+ // the last chunk group may contain the same data with the logs, ignore
such logs in seq file
+ Long lastEndTime =
currentTsFileResource.getEndTimeMap().get(batchInsertPlan.getDeviceId());
+ if (lastEndTime != null && lastEndTime >= batchInsertPlan.getMinTime() &&
+ !acceptDuplication) {
+ return;
+ }
+ Long startTime = tempStartTimeMap.get(batchInsertPlan.getDeviceId());
+ if (startTime == null || startTime > batchInsertPlan.getMinTime()) {
+ tempStartTimeMap.put(batchInsertPlan.getDeviceId(),
batchInsertPlan.getMinTime());
+ }
+ Long endTime = tempEndTimeMap.get(batchInsertPlan.getDeviceId());
+ if (endTime == null || endTime < batchInsertPlan.getMaxTime()) {
+ tempEndTimeMap.put(batchInsertPlan.getDeviceId(),
batchInsertPlan.getMaxTime());
+ }
+ }
+ ArrayList<Integer> index = new ArrayList<>();
+ for (int i = 0; i < batchInsertPlan.getRowCount(); i++) {
+ index.add(i);
}
+ recoverMemTable.insertBatch(batchInsertPlan, index);
}
private void replayInsert(InsertPlan insertPlan) throws
QueryProcessorException {
if (currentTsFileResource != null) {
// the last chunk group may contain the same data with the logs, ignore
such logs in seq file
Long lastEndTime =
currentTsFileResource.getEndTimeMap().get(insertPlan.getDeviceId());
- if ( lastEndTime != null && lastEndTime >= insertPlan.getTime() &&
+ if (lastEndTime != null && lastEndTime >= insertPlan.getTime() &&
!acceptDuplication) {
return;
}
