private void extractTree(final TreeMessage message, final MqttClientSession session) {
    try {
      InsertRowStatement statement = new InsertRowStatement();
      statement.setDevicePath(
          DataNodeDevicePathCache.getInstance().getPartialPath(message.getDevice()));
      TimestampPrecisionUtils.checkTimestampPrecision(message.getTimestamp());
      statement.setTime(message.getTimestamp());
      statement.setMeasurements(message.getMeasurements().toArray(new String[0]));
      if (message.getDataTypes() == null) {
        statement.setDataTypes(new TSDataType[message.getMeasurements().size()]);
        statement.setValues(message.getValues().toArray(new Object[0]));
        statement.setNeedInferType(true);
      } else {
        List<TSDataType> dataTypes = message.getDataTypes();
        List<String> values = message.getValues();
        Object[] inferredValues = new Object[values.size()];
        for (int i = 0; i < values.size(); ++i) {
          inferredValues[i] =
              values.get(i) == null
                  ? null
                  : CommonUtils.parseValue(dataTypes.get(i), values.get(i));
        }
        statement.setDataTypes(dataTypes.toArray(new TSDataType[0]));
        statement.setValues(inferredValues);
      }
      statement.setAligned(false);
      final EnrichedEvent event =
          new PipeStatementInsertionEvent(
              pipeName,
              creationTime,
              pipeTaskMeta,
              null,
              null,
              session.getUsername(),
              true,
              false,
              message.getDevice(),
              statement);
      if (!event.increaseReferenceCount(MQTTPublishHandler.class.getName())) {
        LOGGER.warn("The reference count of the event {} cannot be increased, skipping it.", event);
        return;
      }
      pendingQueue.waitedOffer(event);
    } catch (Exception e) {
      LOGGER.warn(
          "meet error when polling mqtt source device {}, measurements {}, at time {}, because {}",
          message.getDevice(),
          message.getMeasurements(),
          message.getTimestamp(),
          e.getMessage(),
          e);
    }
  }
PublishHandler handle the messages from MQTT clients.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java