package com.muyu.common.iotdb.service.impl;

import com.alibaba.fastjson2.JSON;
import com.muyu.common.iotdb.config.IotDBSessionConfig;
import com.muyu.common.iotdb.domain.dto.IotDbRecordAble;
import com.muyu.common.iotdb.domain.dto.MeasurementSchemaValuesDTO;
import com.muyu.common.iotdb.service.IotDBService;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.springframework.stereotype.Service;

import java.lang.reflect.Type;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @Author WangXin
 * @Data 2024/9/28
 * @Description IotDBServiceImpl业务实现层
 * @Version 1.0.0
 */
@Service
@Slf4j
public class IotDBServiceImpl implements IotDBService {

    @Resource
    private IotDBSessionConfig iotDBSessionConfig;

    /**
     * 单设备批量插入数据
     *
     * @param tablet
     */
    @Override
    public void insertTablet(Tablet tablet) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：tablet:[{}]", tablet);
            sessionPool.insertTablet(tablet);
        } catch (Exception e) {
            log.error("IotDBSession insertTablet失败: tablet={},  error={}", tablet, e.getMessage());
        }
    }

    /**
     * 多设备批量插入数据
     *
     * @param tablets
     */
    @Override
    public void insertTablets(Map<String, Tablet> tablets) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：tablets:[{}]", tablets);
            sessionPool.insertTablets(tablets);
        } catch (Exception e) {
            log.error("IotDBSession insertTablets失败: tablets={},  error={}", tablets, e.getMessage());
        }
    }

    /**
     * 单条数据插入（string类型数据项）
     *
     * @param deviceId     设备名（表名）root.ln.wf01.wt01
     * @param time         时间戳
     * @param measurements 数据项列表
     * @param values       数据项对应值列表
     */
    @Override
    public void insertStringRecord(String deviceId, Long time, List<String> measurements, List<String> values) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：device_id:[{}], measurements:[{}], values:[{}]", deviceId, measurements, values);
            sessionPool.insertRecord(deviceId, time, measurements, values);
        } catch (Exception e) {
            log.error("IotDBSession insertRecord失败: deviceId={}, time={}, measurements={}, values={}, error={}",
                    deviceId, time, measurements, values, e.getMessage());
        }
    }

    /**
     * 单条数据插入（不同类型数据项）
     *
     * @param deviceId     设备名（表名）root.ln.wf01.wt01
     * @param time         时间戳
     * @param measurements 数据项列表
     * @param types        数据项对应类型列表
     * @param values       数据项对应值列表
     */
    @Override
    public void insertRecord(String deviceId, Long time, List<String> measurements, List<TSDataType> types, List<Object> values) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：device_id:[{}], measurements:[{}], types:[{}], values:[{}]", deviceId, measurements, types, values);
            sessionPool.insertRecord(deviceId, time, measurements, types, values);
        } catch (Exception e) {
            log.error("IotDBSession insertRecordHasTypes失败: deviceId={}, time={}, measurements={},types={}, values={}, error={}",
                    deviceId, time, measurements, types, values, e.getMessage());
        }
    }


    /**
     * 多个设备多条数据插入（string类型数据项）
     *
     * @param deviceIds        多个设备名（表名）root.ln.wf01.wt01
     * @param times            时间戳的列表
     * @param measurementsList 数据项列表的列表
     * @param valuesList       数据项对应值列表的列表
     */
    @Override
    public void insertStringRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：deviceIds:[{}], measurementsList:[{}], valuesList:[{}]", deviceIds, measurementsList, valuesList);
            sessionPool.insertRecords(deviceIds, times, measurementsList, valuesList);
        } catch (Exception e) {
            log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, valuesList={}, error={}",
                    deviceIds, times, measurementsList, valuesList, e.getMessage());
        }
    }

    /**
     * 多个设备多条数据插入（不同类型数据项）
     *
     * @param deviceIds        多个设备名（表名））root.ln.wf01.wt01
     * @param times            时间戳的列表
     * @param measurementsList 数据项列表的列表
     * @param typesList        数据项对应类型列表的列表
     * @param valuesList       数据项对应值列表的列表
     */
    @Override
    public void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：deviceIds:[{}], measurementsList:[{}], typesList：[{}], valuesList:[{}]", deviceIds, measurementsList, typesList, valuesList);
            sessionPool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList);
        } catch (Exception e) {
            log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}",
                    deviceIds, times, measurementsList, typesList, valuesList, e.getMessage());
        }
    }

    /**
     * 单个设备多条数据插入（string类型数据项）
     *
     * @param deviceId         单个设备名（表名））root.ln.wf01.wt01
     * @param times            时间戳的列表
     * @param measurementsList 数据项列表的列表
     * @param valuesList       数据项对应值列表的列表
     */
    @Override
    public void insertStringRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：deviceId:[{}], measurementsList:[{}], valuesList:[{}]", deviceId, measurementsList, valuesList);
            sessionPool.insertStringRecordsOfOneDevice(deviceId, times, measurementsList, valuesList);
        } catch (Exception e) {
            log.error("IotDBSession insertStringRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, valuesList={}, error={}",
                    deviceId, times, measurementsList, valuesList, e.getMessage());
        }
    }

    /**
     * 单个设备多条数据插入（不同类型数据项）
     *
     * @param deviceId         单个设备名（表名））root.ln.wf01.wt01
     * @param times            时间戳的列表
     * @param measurementsList 数据项列表的列表
     * @param typesList        数据项对应类型列表的列表
     * @param valuesList       数据项对应值列表的列表
     */
    @Override
    public void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据入库：deviceId:[{}], measurementsList:[{}], typesList：[{}], valuesList:[{}]", deviceId, measurementsList, typesList, valuesList);
            sessionPool.insertRecordsOfOneDevice(deviceId, times, measurementsList, typesList, valuesList);
        } catch (Exception e) {
            log.error("IotDBSession insertRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}", deviceId, times, measurementsList, typesList, valuesList, e.getMessage());
        }
    }

    /**
     * 删除数据（删除一个时间序列在某个时间点前或这个时间点的数据）
     *
     * @param path    单个字段 root.ln.wf01.wt01.temperature
     * @param endTime 删除时间点
     */
    @Override
    public void deleteData(String path, Long endTime) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据删除：path:[{}], endTime:[{}]", path, endTime);
            sessionPool.deleteData(path, endTime);
        } catch (Exception e) {
            log.error("IotDBSession deleteData失败: deviceId={}, times={},error={}", path, endTime, e.getMessage());
        }
    }

    /**
     * 删除数据（删除多个时间序列在某个时间点前或这个时间点的数据）
     *
     * @param paths   多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param endTime 删除时间点
     */
    @Override
    public void deleteData(List<String> paths, Long endTime) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb数据删除：paths:[{}], endTime:[{}]", paths, endTime);
            sessionPool.deleteData(paths, endTime);
        } catch (Exception e) {
            log.error("IotDBSession deleteData失败: paths={}, times={},error={}", paths, endTime, e.getMessage());
        }
    }

    /**
     * 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
     *
     * @param paths     多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @param outTime   超时时间
     * @return SessionDataSet  (Time,paths)
     */
    @Override
    public SessionDataSet executeRawDataQuery(List<String> paths, Long startTime, Long endTime, Long outTime) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb数据查询：paths:[{}], startTime：[{}], endTime:[{}],outTime:[{}]", paths, startTime, endTime, outTime);
            sessionDataSetWrapper = sessionPool.executeRawDataQuery(paths, startTime, endTime, outTime);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeRawDataQuery失败: paths={},  startTime：[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
     *
     * @param paths     多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param startTime 开始时间
     * @param endTime   结束时间
     * @param outTime   超时时间
     * @param clazz     返回数据对应的对象（对象属性必须与字段名对应）
     * @param <T>
     * @return
     */
    @Override
    public <T> List<T> executeRawDataQuery(List<String> paths, Long startTime, Long endTime, Long outTime, Class<? extends IotDbRecordAble> clazz) {
        SessionDataSet sessionDataSet = executeRawDataQuery(paths, startTime, endTime, outTime);
        List<String> columnNames = sessionDataSet.getColumnNames();
        List<T> resultEntities = null;
        try {
            resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
        } catch (Exception e) {
            log.error("IotDBSession executeRawDataQuery失败: paths={},  startTime：[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
        }
        return resultEntities;
    }

    /**
     * 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
     *
     * @param paths    多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param lastTime 结束时间
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeLastDataQuery(List<String> paths, Long lastTime) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb数据查询：paths:[{}], lastTime:[{}]", paths, lastTime);
            sessionDataSetWrapper = sessionPool.executeLastDataQuery(paths, lastTime);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeLastDataQuery失败: paths={},  lastTime：[{}], error={}", paths, lastTime, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
     *
     * @param <T>
     * @param paths    多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param lastTime 结束时间
     * @param clazz    返回数据对应的对象（对象属性必须与字段名对应）
     * @return
     */
    @Override
    public <T> List<T> executeLastDataQuery(List<String> paths, Long lastTime, Class<? extends IotDbRecordAble> clazz) {
        SessionDataSet sessionDataSet = executeLastDataQuery(paths, lastTime);
        List<String> columnNames = sessionDataSet.getColumnNames();
        List<T> resultEntities = null;
        try {
            resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
        } catch (Exception e) {
            log.error("IotDBSession executeLastDataQuery失败: paths={},  lastTime：[{}], error={}", paths, lastTime, e.getMessage());
        }
        return resultEntities;
    }

    /**
     * 最新点查询（快速查询单设备下指定序列最新点）
     *
     * @param db               root.ln.wf01
     * @param device           root.ln.wf01.wt01
     * @param sensors          temperature，status(字段名)
     * @param isLegalPathNodes true(避免路径校验)
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb数据查询：db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}]", db, device, sensors, isLegalPathNodes);
            sessionDataSetWrapper = sessionPool.executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * @param db               root.ln.wf01
     * @param device           root.ln.wf01.wt01
     * @param sensors          temperature，status(字段名)
     * @param isLegalPathNodes true(避免路径校验)
     * @param clazz            返回数据对应的对象（对象属性必须与字段名对应）
     * @param <T>
     * @return
     */
    @Override
    public <T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes, Class<? extends IotDbRecordAble> clazz) {
        SessionDataSet sessionDataSet = executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
        List<String> columnNames = sessionDataSet.getColumnNames();
        List<T> resultEntities = null;
        try {
            resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
        } catch (Exception e) {
            log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
        }
        return resultEntities;
    }

    /**
     * 聚合查询
     *
     * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb聚合查询：paths:[{}], aggregations：[{}]", paths, aggregations);
            sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,error={}", paths, aggregations, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * 聚合查询(时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间)
     *
     * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
     * @param startTime    开始时间(包含)
     * @param endTime      结束时间
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, Long startTime, Long endTime) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}]", paths, aggregations, startTime, endTime);
            sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}],error={}", paths, aggregations, startTime, endTime, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * 聚合查询(支持按照时间区间分段查询)
     *
     * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
     * @param startTime    开始时间(包含)
     * @param endTime      结束时间
     * @param interval
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, Long startTime, Long endTime, Long interval) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;
        try {
            log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}] ,interval:[{}]", paths, aggregations, startTime, endTime, interval);
            sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}], interval:[{}], error={}", paths, aggregations, startTime, endTime, interval, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * 聚合查询(支持按照时间区间分段查询)
     *
     * @param paths        多个字段（表名）） root.ln.wf01.wt01.temperature
     * @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
     * @param startTime    开始时间(包含)
     * @param endTime      结束时间
     * @param interval
     * @param slidingStep
     * @return SessionDataSet
     */
    @Override
    public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, Long startTime, Long endTime, Long interval, Long slidingStep) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;

        try {
            log.info("iotdb聚合查询：paths:[{}], aggregations：[{}],startTime：[{}], endTime:[{}] ,interval:[{}], slidingStep:[{}]", paths, aggregations, startTime, endTime, interval, slidingStep);
            sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval, slidingStep);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations：[{}] ,startTime：[{}], endTime:[{}], interval:[{}], slidingStep:[{}] ,error={}", paths, aggregations, startTime, endTime, interval, slidingStep, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * SQL查询
     *
     * @param sql
     * @return
     */
    @Override
    public SessionDataSet executeQueryStatement(String sql) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        SessionDataSetWrapper sessionDataSetWrapper = null;

        try {
            log.info("iotdb SQL查询：sql:[{}]", sql);
            sessionDataSetWrapper = sessionPool.executeQueryStatement(sql);
            return sessionDataSetWrapper.getSessionDataSet();
        } catch (Exception e) {
            log.error("IotDBSession executeQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
        } finally {
            sessionPool.closeResultSet(sessionDataSetWrapper);
        }
        return null;
    }

    /**
     * SQL非查询
     *
     * @param sql
     */
    @Override
    public void executeNonQueryStatement(String sql) {
        SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
        try {
            log.info("iotdb SQL无查询：sql:[{}]", sql);
            sessionPool.executeNonQueryStatement(sql);
        } catch (Exception e) {
            log.error("IotDBSession executeNonQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
        }
    }

    /**
     * 封装处理数据
     *
     * @param sessionDataSet
     * @param titleList
     */
    @SneakyThrows
    @Override
    public List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> titleList) {
        int fetchSize = sessionDataSet.getFetchSize();
        List<Map<String, Object>> resultList = new ArrayList<>();
        titleList.remove("Time");
        if (fetchSize > 0) {
            while (sessionDataSet.hasNext()) {
                Map<String, Object> resultMap = new HashMap<>();
                RowRecord next = sessionDataSet.next();
                List<Field> fields = next.getFields();
                String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
                resultMap.put("time", timeString);
                for (int i = 0; i < fields.size(); i++) {
                    Field field = fields.get(i);
                    if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
                        resultMap.put(splitString(titleList.get(i)), null);
                    } else {
                        resultMap.put(splitString(titleList.get(i)), field.getObjectValue(field.getDataType()).toString());
                    }
                }
                resultList.add(resultMap);
            }
        }
        return resultList;
    }

    /**
     * 封装处理数据（不支持聚合查询）
     *
     * @param sessionDataSet 查询返回的结果集
     * @param titleList      查询返回的结果集内的字段名
     * @param clazz          返回数据对应的对象（对象属性必须与字段名对应）
     * @param <T>
     * @return
     */
    @SneakyThrows
    @Override
    public <T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList, Class<? extends IotDbRecordAble> clazz) {
        int fetchSize = sessionDataSet.getFetchSize();
        List<T> resultList = new ArrayList<>();
        titleList.remove("Time");
        if (fetchSize > 0) {
            while (sessionDataSet.hasNext()) {
                Map<String, Object> resultMap = new HashMap<>();
                RowRecord next = sessionDataSet.next();
                List<Field> fields = next.getFields();
                String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
                resultMap.put("time", timeString);
                if (titleList.stream().anyMatch(str -> str.contains("."))) {
                    for (int i = 0; i < fields.size(); i++) {
                        Field field = fields.get(i);
                        String title = titleList.get(i);
                        if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
                            resultMap.put(splitString(title), null);
                        } else {
                            resultMap.put(splitString(title), field.getObjectValue(field.getDataType()).toString());
                        }
                    }
                } else {
                    Field fieldName = fields.get(0);
                    Field fieldValue = fields.get(1);
                    Field fieldDataType = fields.get(2);
                    if (fieldName.getDataType() != null && fieldName.getObjectValue(fieldName.getDataType()) != null) {
                        String mapKey = fieldName.getObjectValue(fieldName.getDataType()).toString();
                        Object mapValue = convertStringToType(fieldValue.getObjectValue(fieldValue.getDataType()).toString(), fieldDataType.getObjectValue(fieldDataType.getDataType()).toString());
                        resultMap.put(splitString(mapKey), mapValue);
                    }
                }

                String jsonString = JSON.toJSONString(resultMap);
                resultList.add(JSON.parseObject(jsonString, (Type) clazz));
            }
        }
        return resultList;
    }

    /**
     * 分割获取字段名
     *
     * @param str
     * @return 字段名
     */
    public static String splitString(String str) {
        String[] parts = str.split("\\.");
        if (parts.length <= 0) {
            return str;
        } else {
            return parts[parts.length - 1];
        }
    }

    /**
     * 根据数据值和数据类型返回对应数据类型数据
     *
     * @param value    数据值
     * @param typeName 数据类型
     * @return 转换后的数据值
     */
    public static Object convertStringToType(String value, String typeName) {
        String type = typeName.toLowerCase();
        if (type.isEmpty()) {
            return value;
        }
        if ("boolean".equals(type)) {
            return Boolean.parseBoolean(value);
        } else if ("double".equals(type)) {
            return Double.parseDouble(value);
        } else if ("int32".equals(type)) {
            return Integer.parseInt(value);
        } else if ("int64".equals(type)) {
            return Long.parseLong(value);
        } else if ("float".equals(type)) {
            return Float.parseFloat(value);
        } else if ("text".equals(type)) {
            return value;
        } else {
            return value;
        }
    }

    /**
     * 根据对象属性的数据类型返回对应的TSDataType
     *
     * @param type 属性的数据类型
     * @return TSDataType
     */
    public static TSDataType getTsDataTypeByString(String type) {
        String typeName = splitString(type).toLowerCase();
        if ("boolean".equals(typeName)) {
            return TSDataType.BOOLEAN;
        } else if ("double".equals(typeName)) {
            return TSDataType.DOUBLE;
        } else if ("int".equals(typeName) || "integer".equals(typeName)) {
            return TSDataType.INT32;
        } else if ("Long".equals(typeName)) {
            return TSDataType.INT64;
        } else if ("float".equals(typeName)) {
            return TSDataType.FLOAT;
        } else if ("text".equals(typeName)) {
            return TSDataType.TEXT;
        } else if ("string".equals(typeName)) {
            return TSDataType.TEXT;
        } else {
            return TSDataType.UNKNOWN;
        }
    }

    /**
     * 根据对象构建MeasurementSchemas
     *
     * @param obj 对象
     * @return
     */
    @Override
    public List<MeasurementSchema> buildMeasurementSchemas(Object obj) {
        java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
        List<MeasurementSchema> schemaList = Arrays.stream(fields).map(field ->
                new MeasurementSchema(field.getName(),
                        getTsDataTypeByString(
                                field.getType().getName()
                        ))).
                collect(Collectors.toList());
        return schemaList;
    }

    /**
     * 根据对象构建MeasurementSchemaValuesDTO
     *
     * @param obj 对象
     * @return
     */
    @SneakyThrows
    @Override
    public MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object obj) {
        MeasurementSchemaValuesDTO measurementSchemaValuesDTO = new MeasurementSchemaValuesDTO();
        java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
        List<MeasurementSchema> schemaList = new ArrayList<>();
        List<Object> values = new ArrayList<>();
        List<Integer> valuesIsNullIndex = new ArrayList<>();
        int valueIndex = 0;
        for (java.lang.reflect.Field field : fields) {
            MeasurementSchema measurementSchema = new MeasurementSchema(field.getName(), getTsDataTypeByString(field.getType().getName()));
            schemaList.add(measurementSchema);
            Object value = field.get(obj);
            if (value == null) {
                valuesIsNullIndex.add(valueIndex);
            }
            values.add(value);
            valueIndex++;
        }
        measurementSchemaValuesDTO.setSchemaList(schemaList);
        measurementSchemaValuesDTO.setValues(values);
        return measurementSchemaValuesDTO;
    }
}
