This is an automated email from the ASF dual-hosted git repository. jiangtian pushed a commit to branch print_measurement_for_putting_buffer_exceptions in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e8f26c59446a5480902b302be00040b49ea17e54 Author: Tian Jiang <[email protected]> AuthorDate: Fri Jul 4 11:42:29 2025 +0800 Print the measurement name when exception occured during constructing record --- .../iotdb/session/it/IoTDBSessionSimpleIT.java | 10 ++ .../java/org/apache/iotdb/session/Session.java | 17 ++-- .../apache/iotdb/session/util/SessionUtils.java | 103 ++++++++++++--------- .../iotdb/session/util/SessionUtilsTest.java | 26 +++++- .../iotdb/confignode/manager/UDFManager.java | 2 + .../metrics/IoTDBInternalLocalReporter.java | 2 +- 6 files changed, 104 insertions(+), 56 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java index 543a3707c02..1119dac4ae6 100644 --- a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java @@ -79,6 +79,7 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -2200,4 +2201,13 @@ public class IoTDBSessionSimpleIT { } } } + + @Test + public void testInsertWrongTypeRecord() + throws IoTDBConnectionException { + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + assertThrows(ClassCastException.class, + () -> session.insertRecord("root.db1.d1", 0, Collections.singletonList("s1"), Collections.singletonList(TSDataType.INT32), Collections.singletonList(1L))); + } + } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index e1453c8d29d..da3185e51f8 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -1557,7 +1557,7 @@ public class Session implements ISession { request.setPrefixPath(prefixPath); request.setTimestamp(time); request.setMeasurements(measurements); - ByteBuffer buffer = SessionUtils.getValueBuffer(types, values); + ByteBuffer buffer = SessionUtils.getValueBuffer(types, values, measurements); request.setValues(buffer); request.setIsAligned(isAligned); return request; @@ -2498,7 +2498,8 @@ public class Session implements ISession { request.setPrefixPath(prefixPath); request.setTimestamps(times); request.setMeasurementsList(measurementsList); - List<ByteBuffer> buffersList = objectValuesListToByteBufferList(valuesList, typesList); + List<ByteBuffer> buffersList = + objectValuesListToByteBufferList(valuesList, typesList, measurementsList); request.setValuesList(buffersList); request.setIsAligned(isAligned); return request; @@ -2572,11 +2573,14 @@ public class Session implements ISession { } private List<ByteBuffer> objectValuesListToByteBufferList( - List<List<Object>> valuesList, List<List<TSDataType>> typesList) + List<List<Object>> valuesList, + List<List<TSDataType>> typesList, + List<List<String>> measurementsList) throws IoTDBConnectionException { List<ByteBuffer> buffersList = new ArrayList<>(); for (int i = 0; i < valuesList.size(); i++) { - ByteBuffer buffer = SessionUtils.getValueBuffer(typesList.get(i), valuesList.get(i)); + ByteBuffer buffer = + SessionUtils.getValueBuffer(typesList.get(i), valuesList.get(i), measurementsList.get(i)); buffersList.add(buffer); } return buffersList; @@ -2652,7 +2656,8 @@ public class Session implements ISession { request.setTimestamps(times); request.setMeasurementsList(measurementsList); request.setIsAligned(isAligned); - List<ByteBuffer> buffersList = objectValuesListToByteBufferList(valuesList, typesList); + List<ByteBuffer> buffersList = + objectValuesListToByteBufferList(valuesList, typesList, measurementsList); request.setValuesList(buffersList); return request; } @@ -2689,7 +2694,7 @@ public class Session implements ISession { request.addToPrefixPaths(deviceId); request.addToTimestamps(time); request.addToMeasurementsList(measurements); - ByteBuffer buffer = SessionUtils.getValueBuffer(types, values); + ByteBuffer buffer = SessionUtils.getValueBuffer(types, values, measurements); request.addToValuesList(buffer); } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java index 8534b808754..7a43578b975 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java @@ -34,6 +34,8 @@ import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.UnSupportedDataTypeException; import org.apache.tsfile.write.record.Tablet; import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.nio.ByteBuffer; import java.time.LocalDate; @@ -44,6 +46,7 @@ import static org.apache.iotdb.session.Session.MSG_UNSUPPORTED_DATA_TYPE; public class SessionUtils { + private static final Logger LOGGER = LoggerFactory.getLogger(SessionUtils.class); private static final byte TYPE_NULL = -2; private static final int EMPTY_DATE_INT = 10000101; @@ -145,10 +148,11 @@ public class SessionUtils { return valueOccupation; } - public static ByteBuffer getValueBuffer(List<TSDataType> types, List<Object> values) + public static ByteBuffer getValueBuffer( + List<TSDataType> types, List<Object> values, List<String> measurements) throws IoTDBConnectionException { ByteBuffer buffer = ByteBuffer.allocate(SessionUtils.calculateLength(types, values)); - SessionUtils.putValues(types, values, buffer); + SessionUtils.putValues(types, values, buffer, measurements); return buffer; } @@ -204,53 +208,60 @@ public class SessionUtils { * @param buffer buffer to insert * @throws IoTDBConnectionException */ - private static void putValues(List<TSDataType> types, List<Object> values, ByteBuffer buffer) + private static void putValues( + List<TSDataType> types, List<Object> values, ByteBuffer buffer, List<String> measurements) throws IoTDBConnectionException { for (int i = 0; i < values.size(); i++) { - if (values.get(i) == null) { - ReadWriteIOUtils.write(TYPE_NULL, buffer); - continue; - } - ReadWriteIOUtils.write(types.get(i), buffer); - switch (types.get(i)) { - case BOOLEAN: - ReadWriteIOUtils.write((Boolean) values.get(i), buffer); - break; - case INT32: - ReadWriteIOUtils.write((Integer) values.get(i), buffer); - break; - case DATE: - ReadWriteIOUtils.write( - DateUtils.parseDateExpressionToInt((LocalDate) values.get(i)), buffer); - break; - case INT64: - case TIMESTAMP: - ReadWriteIOUtils.write((Long) values.get(i), buffer); - break; - case FLOAT: - ReadWriteIOUtils.write((Float) values.get(i), buffer); - break; - case DOUBLE: - ReadWriteIOUtils.write((Double) values.get(i), buffer); - break; - case TEXT: - case STRING: - byte[] bytes; - if (values.get(i) instanceof Binary) { + try { + if (values.get(i) == null) { + ReadWriteIOUtils.write(TYPE_NULL, buffer); + continue; + } + ReadWriteIOUtils.write(types.get(i), buffer); + switch (types.get(i)) { + case BOOLEAN: + ReadWriteIOUtils.write((Boolean) values.get(i), buffer); + break; + case INT32: + ReadWriteIOUtils.write((Integer) values.get(i), buffer); + break; + case DATE: + ReadWriteIOUtils.write( + DateUtils.parseDateExpressionToInt((LocalDate) values.get(i)), buffer); + break; + case INT64: + case TIMESTAMP: + ReadWriteIOUtils.write((Long) values.get(i), buffer); + break; + case FLOAT: + ReadWriteIOUtils.write((Float) values.get(i), buffer); + break; + case DOUBLE: + ReadWriteIOUtils.write((Double) values.get(i), buffer); + break; + case TEXT: + case STRING: + byte[] bytes; + if (values.get(i) instanceof Binary) { + bytes = ((Binary) values.get(i)).getValues(); + } else { + bytes = ((String) values.get(i)).getBytes(TSFileConfig.STRING_CHARSET); + } + ReadWriteIOUtils.write(bytes.length, buffer); + buffer.put(bytes); + break; + case BLOB: bytes = ((Binary) values.get(i)).getValues(); - } else { - bytes = ((String) values.get(i)).getBytes(TSFileConfig.STRING_CHARSET); - } - ReadWriteIOUtils.write(bytes.length, buffer); - buffer.put(bytes); - break; - case BLOB: - bytes = ((Binary) values.get(i)).getValues(); - ReadWriteIOUtils.write(bytes.length, buffer); - buffer.put(bytes); - break; - default: - throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + types.get(i)); + ReadWriteIOUtils.write(bytes.length, buffer); + buffer.put(bytes); + break; + default: + throw new IoTDBConnectionException(MSG_UNSUPPORTED_DATA_TYPE + types.get(i)); + } + } catch (Throwable e) { + LOGGER.error( + "Cannot put values for measurement {}, type={}", measurements.get(i), types.get(i), e); + throw e; } } buffer.flip(); diff --git a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java index 5818a89625e..ffdc5835dfc 100644 --- a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java +++ b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java @@ -40,6 +40,8 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import static org.junit.Assert.assertThrows; + public class SessionUtilsTest { @Test @@ -125,19 +127,20 @@ public class SessionUtilsTest { TSDataType.DOUBLE, TSDataType.TEXT, TSDataType.BOOLEAN); - ByteBuffer timeBuffer = SessionUtils.getValueBuffer(typeList, valueList); + List<String> measurements = Arrays.asList("s1", "s2", "s3", "s4", "s5", "s6"); + ByteBuffer timeBuffer = SessionUtils.getValueBuffer(typeList, valueList, measurements); Assert.assertNotNull(timeBuffer); valueList = new ArrayList<>(); valueList.add(null); typeList = Collections.singletonList(TSDataType.INT32); - timeBuffer = SessionUtils.getValueBuffer(typeList, valueList); + timeBuffer = SessionUtils.getValueBuffer(typeList, valueList, measurements); Assert.assertNotNull(timeBuffer); valueList = Collections.singletonList(false); typeList = Collections.singletonList(TSDataType.UNKNOWN); try { - SessionUtils.getValueBuffer(typeList, valueList); + SessionUtils.getValueBuffer(typeList, valueList, measurements); } catch (Exception e) { Assert.assertTrue(e instanceof IoTDBConnectionException); } @@ -204,6 +207,23 @@ public class SessionUtilsTest { Assert.assertNotNull(timeBuffer); } + @Test + public void testGetValueBufferWithWrongType() { + List<Object> valueList = Arrays.asList(12L, 13, 1.2, 0.707f, false, "false"); + List<TSDataType> typeList = + Arrays.asList( + TSDataType.INT32, + TSDataType.INT64, + TSDataType.FLOAT, + TSDataType.DOUBLE, + TSDataType.TEXT, + TSDataType.BOOLEAN); + List<String> measurements = Arrays.asList("s1", "s2", "s3", "s4", "s5", "s6"); + assertThrows( + ClassCastException.class, + () -> SessionUtils.getValueBuffer(typeList, valueList, measurements)); + } + @Test public void testParseSeedNodeUrls() { List<String> nodeUrls = Collections.singletonList("127.0.0.1:1234"); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java index 6c6477e2418..4eef7ea83bf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/UDFManager.java @@ -72,6 +72,8 @@ public class UDFManager { .getConfigNodeRatisConsensusLogAppenderBufferSize() - IoTDBConstant.RAFT_LOG_BASIC_SIZE; + + public UDFManager(ConfigManager configManager, UDFInfo udfInfo) { this.configManager = configManager; this.udfInfo = udfInfo; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java index e6d37e49b9c..083d5fcd06c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/metrics/IoTDBInternalLocalReporter.java @@ -191,7 +191,7 @@ public class IoTDBInternalLocalReporter extends IoTDBInternalReporter { types.add(inferType(value)); values.add(value); } - ByteBuffer buffer = SessionUtils.getValueBuffer(types, values); + ByteBuffer buffer = SessionUtils.getValueBuffer(types, values, measurements); request.setPrefixPath(prefix); request.setTimestamp(time);
