private InsertTabletStatement constructInsertTabletStatement(TableMessage message)
      throws IllegalPathException {
    InsertTabletStatement statement = new InsertTabletStatement();
    statement.setDevicePath(
        DataNodeDevicePathCache.getInstance().getPartialPath(message.getTable()));
    List<String> measurements =
        Stream.of(message.getFields(), message.getTagKeys(), message.getAttributeKeys())
            .flatMap(List::stream)
            .collect(Collectors.toList());
    statement.setMeasurements(measurements.toArray(new String[0]));
    long[] timestamps = new long[] {message.getTimestamp()};
    statement.setTimes(timestamps);
    int columnSize = measurements.size();
    int rowSize = 1;

    BitMap[] bitMaps = new BitMap[columnSize];
    Object[] columns =
        Stream.of(message.getValues(), message.getTagValues(), message.getAttributeValues())
            .flatMap(List::stream)
            .toArray(Object[]::new);
    statement.setColumns(columns);
    statement.setBitMaps(bitMaps);
    statement.setRowCount(rowSize);
    statement.setAligned(false);
    statement.setWriteToTable(true);
    TSDataType[] dataTypes = new TSDataType[measurements.size()];
    TsTableColumnCategory[] columnCategories = new TsTableColumnCategory[measurements.size()];
    for (int i = 0; i < message.getFields().size(); i++) {
      dataTypes[i] = message.getDataTypes().get(i);
      columnCategories[i] = TsTableColumnCategory.FIELD;
    }
    for (int i = message.getFields().size();
        i < message.getFields().size() + message.getTagKeys().size();
        i++) {
      dataTypes[i] = TSDataType.STRING;
      columnCategories[i] = TsTableColumnCategory.TAG;
    }
    for (int i = message.getFields().size() + message.getTagKeys().size();
        i
            < message.getFields().size()
                + message.getTagKeys().size()
                + message.getAttributeKeys().size();
        i++) {
      dataTypes[i] = TSDataType.STRING;
      columnCategories[i] = TsTableColumnCategory.ATTRIBUTE;
    }
    statement.setDataTypes(dataTypes);
    statement.setColumnCategories(columnCategories);

    return statement;
  }
PublishHandler handle the messages from MQTT clients.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java