Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -1379,6 +1379,68 @@ public void testInsertAllNullRow() throws SQLException {
}
}

@Test
public void testRejectedOnlyFieldStillWritesTime() throws Exception {
try (ITableSession session = EnvFactory.getEnv().getTableSessionConnection()) {
session.executeNonQueryStatement("use \"test\"");
session.executeNonQueryStatement(
"create table rejected_only_field_time (tag1 string tag, s1 int64 field)");

Tablet tablet =
new Tablet(
"rejected_only_field_time",
Arrays.asList("tag1", "s1"),
Arrays.asList(TSDataType.STRING, TSDataType.BOOLEAN),
Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
2);
for (int i = 0; i < 2; i++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, i + 1);
tablet.addValue("tag1", row, "d" + i);
tablet.addValue("s1", row, true);
}

StatementExecutionException error =
assertThrows(StatementExecutionException.class, () -> session.insert(tablet));
assertTrue(error.getMessage().contains("s1"));

Tablet withoutTag =
new Tablet(
"rejected_only_field_time",
Arrays.asList("s1"),
Arrays.asList(TSDataType.BOOLEAN),
Arrays.asList(ColumnCategory.FIELD),
1);
withoutTag.addTimestamp(0, 3);
withoutTag.addValue("s1", 0, true);
error = assertThrows(StatementExecutionException.class, () -> session.insert(withoutTag));
assertTrue(error.getMessage().contains("s1"));
assertRejectedFieldTimeRows(session);

session.executeNonQueryStatement("flush");
assertRejectedFieldTimeRows(session);
}
}

private static void assertRejectedFieldTimeRows(ITableSession session)
throws IoTDBConnectionException, StatementExecutionException {
SessionDataSet dataSet =
session.executeQueryStatement(
"select time, tag1, s1 from rejected_only_field_time order by time");
for (int i = 0; i < 3; i++) {
assertTrue(dataSet.hasNext());
RowRecord row = dataSet.next();
assertEquals(i + 1, row.getFields().get(0).getLongV());
if (i < 2) {
assertEquals("d" + i, row.getFields().get(1).getStringValue());
} else {
assertNull(row.getFields().get(1).getObjectValue(TSDataType.STRING));
}
assertNull(row.getFields().get(2).getObjectValue(TSDataType.INT64));
}
assertFalse(dataSet.hasNext());
}

@Test
public void testInsertWithChangedSchema() throws SQLException, IOException {
SimpleEnv simpleEnv = new SimpleEnv();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -437,16 +437,19 @@ public void partialInsertSQLTest() throws IoTDBConnectionException, StatementExe

SessionDataSet dataSet =
session.executeQueryStatement("select * from partial_insert order by time");
long[] timestamps = new long[] {10000, 20000, 40000, 50000};
Boolean[] values = new Boolean[] {true, false, false, true};
// Partial insertion retains timestamps even when the only FIELD value is rejected.
long[] timestamps =
new long[] {10000, 20000, 30000, 35000, 40000, 50000, 60000, 70000, 80000, 90000};
Boolean[] values =
new Boolean[] {true, false, null, null, false, true, null, null, null, null};
int cnt = 0;
while (dataSet.hasNext()) {
RowRecord rec = dataSet.next();
assertEquals(timestamps[cnt], rec.getFields().get(0).getLongV());
assertEquals(values[cnt], rec.getFields().get(1).getBoolV());
assertEquals(values[cnt], rec.getFields().get(1).getObjectValue(TSDataType.BOOLEAN));
cnt++;
}
assertEquals(4, cnt);
assertEquals(timestamps.length, cnt);

} finally {
try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
Expand Down Expand Up @@ -757,14 +760,23 @@ public void insertAllFieldDataTypeMismatchTest()
e.getMessage());
}
tablet.reset();
int cnt = 0;
SessionDataSet dataSet =
session.executeQueryStatement("select * from field_wrong_type order by time");
while (dataSet.hasNext()) {
dataSet.next();
cnt++;
for (int phase = 0; phase < 2; phase++) {
try (SessionDataSet dataSet =
session.executeQueryStatement("select * from field_wrong_type order by time")) {
for (int row = 0; row < 10; row++) {
assertTrue(dataSet.hasNext());
RowRecord record = dataSet.next();
assertEquals(row, record.getFields().get(0).getLongV());
assertEquals("tag:" + row, record.getFields().get(1).getStringValue());
assertNull(record.getFields().get(2).getDataType());
assertNull(record.getFields().get(3).getDataType());
}
assertFalse(dataSet.hasNext());
}
if (phase == 0) {
session.executeNonQueryStatement("flush");
}
}
assertEquals(0, cnt);
}
}

Expand Down Expand Up @@ -1500,9 +1512,17 @@ private void testOneCastWithTablet(
e.getMessage());
}
// time, tag1, m1
SessionDataSet dataSet =
session.executeQueryStatement("select * from table" + testNum + " order by time");
assertFalse(dataSet.hasNext());
try (SessionDataSet dataSet =
session.executeQueryStatement("select * from table" + testNum + " order by time")) {
for (int row = 0; row < 2; row++) {
assertTrue(dataSet.hasNext());
RowRecord record = dataSet.next();
assertEquals(row, record.getFields().get(0).getLongV());
assertEquals("d1", record.getFields().get(1).getStringValue());
assertNull(record.getFields().get(2).getDataType());
}
assertFalse(dataSet.hasNext());
}
} else {
// cannot cast, expect an exception
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3007,6 +3007,7 @@ private void insertRelationalTabletByGroup(Map<SessionConnection, Tablet> relati
.collect(Collectors.toList());

StringBuilder errMsgBuilder = new StringBuilder();
List<Throwable> errors = new ArrayList<>();
for (CompletableFuture<Void> completableFuture : completableFutures) {
try {
completableFuture.join();
Expand All @@ -3016,6 +3017,7 @@ private void insertRelationalTabletByGroup(Map<SessionConnection, Tablet> relati
if (cause instanceof IoTDBConnectionException) {
throw (IoTDBConnectionException) cause;
} else {
errors.add(cause);
if (errMsgBuilder.length() > 0) {
errMsgBuilder.append(";");
}
Expand All @@ -3024,7 +3026,10 @@ private void insertRelationalTabletByGroup(Map<SessionConnection, Tablet> relati
}
}
if (errMsgBuilder.length() > 0) {
throw new StatementExecutionException(errMsgBuilder.toString());
StatementExecutionException exception =
new StatementExecutionException(errMsgBuilder.toString());
errors.forEach(exception::addSuppressed);
throw exception;
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,10 +20,12 @@
package org.apache.iotdb.session;

import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.isession.SessionConfig;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.RedirectException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsOfOneDeviceReq;
import org.apache.iotdb.service.rpc.thrift.TSInsertRecordsReq;
Expand All @@ -44,10 +46,12 @@

import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
Expand Down Expand Up @@ -705,6 +709,79 @@ public void testInsertRelationalTablet()
session.close();
}

@Test
public void testGroupedRelationalTabletPreservesStatementFailures() throws Exception {
AtomicBoolean failWrites = new AtomicBoolean();
session =
new MockSession("127.0.0.1", 55560, true, "table") {
@Override
public SessionConnection constructSessionConnection(
Session owner, TEndPoint endpoint, ZoneId zoneId) {
return new MockSessionConnection(owner, endpoint, zoneId) {
@Override
protected void insertTablet(TSInsertTabletReq request)
throws RedirectException, IoTDBConnectionException, StatementExecutionException {
if (failWrites.get()) {
TSStatusCode code =
endpoint.getPort() == 55560
? TSStatusCode.METADATA_ERROR
: TSStatusCode.DATA_TYPE_MISMATCH;
throw new StatementExecutionException(
new TSStatus(code.getStatusCode()).setMessage("incompatible field"));
}
super.insertTablet(request);
}
};
}
};
session.open();
try {
Tablet tablet =
new Tablet(
"table1",
Arrays.asList("id", "field"),
Arrays.asList(TSDataType.STRING, TSDataType.INT64),
Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
50);
for (int i = 0; i < 50; i++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, row);
tablet.addValue("id", row, "id" + (row % 4));
tablet.addValue("field", row, 1L);
}
session.insertRelationalTablet(tablet);
assertEquals(4, session.tableModelDeviceIdToEndpoint.size());

failWrites.set(true);
tablet.reset();
for (int i = 0; i < 4; i++) {
int row = tablet.getRowSize();
tablet.addTimestamp(row, row);
tablet.addValue("id", row, "id" + row);
tablet.addValue("field", row, 1L);
}
StatementExecutionException failure =
Assert.assertThrows(
StatementExecutionException.class, () -> session.insertRelationalTablet(tablet));
assertEquals(4, failure.getSuppressed().length);
long metadataFailures = 0;
long typeFailures = 0;
for (Throwable cause : failure.getSuppressed()) {
Assert.assertTrue(cause instanceof StatementExecutionException);
int code = ((StatementExecutionException) cause).getStatusCode();
if (code == TSStatusCode.METADATA_ERROR.getStatusCode()) {
metadataFailures++;
} else if (code == TSStatusCode.DATA_TYPE_MISMATCH.getStatusCode()) {
typeFailures++;
}
}
assertEquals(1, metadataFailures);
assertEquals(3, typeFailures);
} finally {
session.close();
}
}

@Test
public void testInsertRecordsWithSessionBroken() throws StatementExecutionException {
// without leader cache
Expand Down Expand Up @@ -1281,7 +1358,7 @@ protected void insertRecordsOfOneDevice(TSInsertRecordsOfOneDeviceReq request)

@Override
protected void insertTablet(TSInsertTabletReq request)
throws RedirectException, IoTDBConnectionException {
throws RedirectException, IoTDBConnectionException, StatementExecutionException {
if (isConnectionBroken()) {
throw ioTDBConnectionException;
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -446,6 +446,11 @@ public boolean allMeasurementFailed() {
return measurements == null || !hasValidMeasurements();
}

public boolean shouldSkipWrite() {
// Table rows remain writable through their time column after every FIELD is rejected.
return allMeasurementFailed() && columnCategories == null;
}

// endregion

// region progress index
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1559,7 +1559,7 @@ private int findLastInsertTabletIndexToMark(final InsertMultiTabletsNode insertM
for (int i = insertMultiTabletsNode.getInsertTabletNodeList().size() - 1; i >= 0; i--) {
final InsertTabletNode insertTabletNode =
insertMultiTabletsNode.getInsertTabletNodeList().get(i);
if (insertTabletNode.getRowCount() <= 0 || insertTabletNode.allMeasurementFailed()) {
if (insertTabletNode.getRowCount() <= 0 || insertTabletNode.shouldSkipWrite()) {
continue;
}
if (!insertTabletNode.shouldCheckTTL()) {
Expand Down Expand Up @@ -1609,7 +1609,7 @@ private boolean insertTabletToTsFileProcessor(
boolean noFailure,
long[] infoForMetrics)
throws DataTypeInconsistentException {
if (insertTabletNode.allMeasurementFailed()) {
if (insertTabletNode.shouldSkipWrite()) {
if (logger.isDebugEnabled()) {
logger.debug(
StorageEngineMessages.STORAGE_LOG_WON_T_INSERT_TABLET_BECAUSE_C2DC8032,
Expand Down Expand Up @@ -1817,7 +1817,7 @@ private void tryToUpdateInsertTabletLastCache(
private TsFileProcessor insertToTsFileProcessor(
InsertRowNode insertRowNode, boolean sequence, long timePartitionId)
throws WriteProcessException {
if (insertRowNode.allMeasurementFailed()) {
if (insertRowNode.shouldSkipWrite()) {
return null;
}
TsFileProcessor tsFileProcessor = getOrCreateTsFileProcessor(timePartitionId, sequence);
Expand Down Expand Up @@ -1846,7 +1846,7 @@ private List<InsertRowNode> insertToTsFileProcessors(
Map<TsFileProcessor, InsertRowsNode> tsFileProcessorMap = new HashMap<>();
for (int i = 0; i < areSequence.length; i++) {
InsertRowNode insertRowNode = insertRowsNode.getInsertRowNodeList().get(i);
if (insertRowNode.allMeasurementFailed()) {
if (insertRowNode.shouldSkipWrite()) {
continue;
}
final TsFileProcessor tsFileProcessor;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -266,7 +266,7 @@ public int insertAlignedRow(InsertRowNode insertRowNode) {
dataTypes.add(schema.getType());
writableValues[i] = value;
}
if (schemaList.isEmpty()) {
if (schemaList.isEmpty() && insertRowNode.getColumnCategories() == null) {
return 0;
}
memSize +=
Expand Down Expand Up @@ -497,7 +497,7 @@ public void writeAlignedTablet(
}
}
}
if (schemaList.isEmpty()) {
if (schemaList.isEmpty() && insertTabletNode.getColumnCategories() == null) {
return;
}
int splitStart = start;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ void redoDelete(RelationalDeleteDataNode node) throws IOException {
}

void redoInsert(InsertNode node) throws WriteProcessException {
if (!node.hasValidMeasurements()) {
if (node.shouldSkipWrite()) {
return;
}
boolean isSingleDevice = !(node instanceof RelationalInsertTabletNode);
Expand Down Expand Up @@ -155,7 +155,7 @@ void redoInsert(InsertNode node) throws WriteProcessException {
void redoInsertRows(InsertRowsNode insertRowsNode) {
int pointsInserted = 0;
for (InsertRowNode node : insertRowsNode.getInsertRowNodeList()) {
if (!node.hasValidMeasurements()) {
if (node.shouldSkipWrite()) {
continue;
}
if (tsFileResource != null) {
Expand Down
Loading
Loading