JackieTien97 commented on code in PR #17027:
URL: https://github.com/apache/iotdb/pull/17027#discussion_r2758226722
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java:
##########
@@ -47,80 +48,70 @@ public boolean isNull() {
}
}
- private PreparedParameterSerializer() {}
+ private PreparedParameterSerde() {}
/** Serialize parameters to binary format. */
public static ByteBuffer serialize(Object[] values, int[] jdbcTypes, int
count) {
- try {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(baos);
-
- dos.writeInt(count);
+ try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
Review Comment:
Using `PublicBAOS` instead of `ByteArrayOutputStream`, and then you can use
`ByteBuffer.wrap(PublicBAOS.getBuf(), 0, PublicBAOS.size())` to avoid copying
in `.toByteArray()`.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java:
##########
@@ -1637,31 +1616,48 @@ public TSStatus
deallocatePreparedStatement(TSDeallocatePreparedReq req) {
}
}
- private Literal convertToLiteral(DeserializedParam param) {
+ private Pair<Literal, String> convertToLiteralWithString(DeserializedParam
param) {
if (param.isNull()) {
- return new NullLiteral();
+ return new Pair<>(new NullLiteral(), "NULL");
}
switch (param.type) {
case BOOLEAN:
- return new BooleanLiteral((Boolean) param.value ? "true" : "false");
+ String boolStr = (boolean) param.value ? "true" : "false";
+ return new Pair<>(new BooleanLiteral(boolStr), boolStr);
case INT32:
case INT64:
- return new LongLiteral(String.valueOf(param.value));
+ String numStr = String.valueOf(param.value);
+ return new Pair<>(new LongLiteral(numStr), numStr);
case FLOAT:
- return new DoubleLiteral((Float) param.value);
+ String floatStr = String.valueOf(param.value);
+ return new Pair<>(new DoubleLiteral((Float) param.value), floatStr);
case DOUBLE:
- return new DoubleLiteral((Double) param.value);
+ String doubleStr = String.valueOf(param.value);
+ return new Pair<>(new DoubleLiteral((Double) param.value), doubleStr);
case TEXT:
case STRING:
- return new StringLiteral((String) param.value);
+ String strVal = (String) param.value;
+ // Escape single quotes for SQL
+ String escapedStr = "'" + strVal.replace("'", "''") + "'";
+ return new Pair<>(new StringLiteral(strVal), escapedStr);
case BLOB:
- return new BinaryLiteral((byte[]) param.value);
+ byte[] bytes = (byte[]) param.value;
+ String hexStr = "X'" + bytesToHex(bytes) + "'";
+ return new Pair<>(new BinaryLiteral(bytes), hexStr);
default:
throw new IllegalArgumentException("Unknown parameter type: " +
param.type);
}
}
+ private static String bytesToHex(byte[] bytes) {
+ StringBuilder sb = new StringBuilder(bytes.length * 2);
+ for (byte b : bytes) {
+ sb.append(String.format("%02X", b));
+ }
+ return sb.toString();
+ }
Review Comment:
you can define it in Serde class, then you can call it here and
`IoTDBTablePreparedStatement`.
##########
iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/stmt/PreparedParameterSerde.java:
##########
@@ -47,80 +48,70 @@ public boolean isNull() {
}
}
- private PreparedParameterSerializer() {}
+ private PreparedParameterSerde() {}
/** Serialize parameters to binary format. */
public static ByteBuffer serialize(Object[] values, int[] jdbcTypes, int
count) {
- try {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- DataOutputStream dos = new DataOutputStream(baos);
-
- dos.writeInt(count);
+ try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
+ ReadWriteIOUtils.write(count, outputStream);
for (int i = 0; i < count; i++) {
- serializeParameter(dos, values[i], jdbcTypes[i]);
+ serializeParameter(outputStream, values[i], jdbcTypes[i]);
}
-
- dos.flush();
- return ByteBuffer.wrap(baos.toByteArray());
+ return ByteBuffer.wrap(outputStream.toByteArray());
} catch (IOException e) {
- throw new RuntimeException("Failed to serialize parameters", e);
+ // Should not happen with ByteArrayOutputStream
+ throw new IllegalStateException("Failed to serialize parameters", e);
}
}
- private static void serializeParameter(DataOutputStream dos, Object value,
int jdbcType)
+ private static void serializeParameter(OutputStream outputStream, Object
value, int jdbcType)
throws IOException {
if (value == null || jdbcType == Types.NULL) {
- dos.writeByte(TSDataType.UNKNOWN.serialize());
+ ReadWriteIOUtils.write(TSDataType.UNKNOWN, outputStream);
return;
}
switch (jdbcType) {
case Types.BOOLEAN:
- dos.writeByte(TSDataType.BOOLEAN.serialize());
- dos.writeByte((Boolean) value ? 1 : 0);
+ ReadWriteIOUtils.write(TSDataType.BOOLEAN, outputStream);
+ ReadWriteIOUtils.write((Boolean) value, outputStream);
Review Comment:
ReadWriteIOUtils.write((boolean) value, outputStream);
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java:
##########
@@ -1542,80 +1554,47 @@ public TSPrepareResp prepareStatement(TSPrepareReq req)
{
@Override
public TSExecuteStatementResp executePreparedStatement(TSExecutePreparedReq
req) {
- boolean finished = false;
- long queryId = Long.MIN_VALUE;
IClientSession clientSession =
SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
- if (!SESSION_MANAGER.checkLogin(clientSession)) {
- return RpcUtils.getTSExecuteStatementResp(getNotLoggedInStatus());
- }
-
- long startTime = System.nanoTime();
- Throwable t = null;
try {
String statementName = req.getStatementName();
+ // Deserialize parameters and build Execute statement
List<DeserializedParam> rawParams =
-
PreparedParameterSerializer.deserialize(ByteBuffer.wrap(req.getParameters()));
+
PreparedParameterSerde.deserialize(ByteBuffer.wrap(req.getParameters()));
List<Literal> parameters = new ArrayList<>(rawParams.size());
+ List<String> paramStrings = new ArrayList<>(rawParams.size());
for (DeserializedParam param : rawParams) {
- parameters.add(convertToLiteral(param));
+ Pair<Literal, String> literalAndString =
convertToLiteralWithString(param);
+ parameters.add(literalAndString.left);
+ paramStrings.add(literalAndString.right);
}
-
Execute executeStatement = new Execute(new Identifier(statementName),
parameters);
- queryId = SESSION_MANAGER.requestQueryId(clientSession,
req.getStatementId());
+ String fullStatement =
+ paramStrings.isEmpty()
+ ? "EXECUTE " + statementName
+ : "EXECUTE " + statementName + " USING " + String.join(", ",
paramStrings);
- long timeout = req.isSetTimeout() ? req.getTimeout() :
config.getQueryTimeoutThreshold();
- ExecutionResult result =
- COORDINATOR.executeForTableModel(
- executeStatement,
- relationSqlParser,
- clientSession,
- queryId,
- SESSION_MANAGER.getSessionInfo(clientSession),
- "EXECUTE " + statementName,
- metadata,
- timeout,
- true);
-
- if (result.status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()
- && result.status.code !=
TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) {
- finished = true;
- return RpcUtils.getTSExecuteStatementResp(result.status);
+ // Build a compatible TSExecuteStatementReq
+ TSExecuteStatementReq executeReq = new TSExecuteStatementReq();
+ executeReq.setSessionId(clientSession.getId());
+ executeReq.setStatement(fullStatement);
+ executeReq.setStatementId(req.getStatementId());
+ if (req.isSetFetchSize()) {
+ executeReq.setFetchSize(req.getFetchSize());
}
-
- IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId);
-
- try (SetThreadName threadName = new
SetThreadName(result.queryId.getId())) {
- TSExecuteStatementResp resp;
- if (queryExecution != null && queryExecution.isQuery()) {
- resp = createResponse(queryExecution.getDatasetHeader(), queryId);
- resp.setStatus(result.status);
- int fetchSize =
- req.isSetFetchSize() ? req.getFetchSize() :
config.getThriftMaxFrameSize();
- finished = setResultForPrepared.apply(resp, queryExecution,
fetchSize);
- resp.setMoreData(!finished);
- } else {
- finished = true;
- resp = RpcUtils.getTSExecuteStatementResp(result.status);
- }
- return resp;
+ if (req.isSetTimeout()) {
+ executeReq.setTimeout(req.getTimeout());
}
+
+ return executeStatementInternal(executeReq, setResultForPrepared, () ->
executeStatement);
Review Comment:
not right, you shouldn't directly reuse current `executeStatementInternal`,
you need to do some changes to it.
For example, you should always do the login check at the very fisrt of each
rpc method, otherwise it may have DoS risks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
