-
Notifications
You must be signed in to change notification settings - Fork 22
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #125 from thulab/modify-for-hadoop-connector
Modify the tsfile for hadoop connector
- Loading branch information
Showing
8 changed files
with
257 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
125 changes: 125 additions & 0 deletions
125
src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/query/HadoopQueryEngine.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package cn.edu.tsinghua.tsfile.timeseries.read.query; | ||
|
|
||
| import cn.edu.tsinghua.tsfile.common.exception.ProcessorException; | ||
| import cn.edu.tsinghua.tsfile.common.utils.ITsRandomAccessFileReader; | ||
| import cn.edu.tsinghua.tsfile.file.metadata.RowGroupMetaData; | ||
| import cn.edu.tsinghua.tsfile.file.metadata.TimeSeriesChunkMetaData; | ||
| import cn.edu.tsinghua.tsfile.timeseries.filter.definition.CrossSeriesFilterExpression; | ||
| import cn.edu.tsinghua.tsfile.timeseries.filter.definition.FilterExpression; | ||
| import cn.edu.tsinghua.tsfile.timeseries.filter.definition.SingleSeriesFilterExpression; | ||
| import cn.edu.tsinghua.tsfile.timeseries.read.RecordReader; | ||
| import cn.edu.tsinghua.tsfile.timeseries.read.support.Path; | ||
| import cn.edu.tsinghua.tsfile.timeseries.write.desc.MeasurementDescriptor; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
|
|
||
| public class HadoopQueryEngine extends QueryEngine { | ||
|
|
||
| private static final String SEPARATOR_DEVIDE_SERIES = "."; | ||
| private List<RowGroupMetaData> rowGroupMetaDataList; | ||
|
|
||
| public HadoopQueryEngine(ITsRandomAccessFileReader raf, List<RowGroupMetaData> rowGroupMetaDataList) throws IOException { | ||
| super(raf, rowGroupMetaDataList); | ||
| this.rowGroupMetaDataList = rowGroupMetaDataList; | ||
| } | ||
|
|
||
| private List<String> initDeviceIdList() { | ||
| Set<String> deviceIdSet = new HashSet<>(); | ||
| for (RowGroupMetaData rowGroupMetaData : rowGroupMetaDataList) { | ||
| deviceIdSet.add(rowGroupMetaData.getDeltaObjectID()); | ||
| } | ||
| return new ArrayList<>(deviceIdSet); | ||
| } | ||
|
|
||
| private List<String> initSensorIdList(){ | ||
| Set<String> sensorIdSet = new HashSet<>(); | ||
| for(RowGroupMetaData rowGroupMetaData : rowGroupMetaDataList) { | ||
| for(TimeSeriesChunkMetaData timeSeriesChunkMetaData : rowGroupMetaData.getTimeSeriesChunkMetaDataList()){ | ||
| sensorIdSet.add(timeSeriesChunkMetaData.getProperties().getMeasurementUID()); | ||
| } | ||
| } | ||
| return new ArrayList<>(sensorIdSet); | ||
| } | ||
|
|
||
| public QueryDataSet queryWithSpecificRowGroups(List<String> deviceIdList, List<String> sensorIdList, FilterExpression timeFilter, FilterExpression freqFilter, FilterExpression valueFilter) throws IOException{ | ||
| if(deviceIdList == null)deviceIdList = initDeviceIdList(); | ||
| if(sensorIdList == null)sensorIdList = initSensorIdList(); | ||
|
|
||
| List<Path> paths = new ArrayList<>(); | ||
| for(String deviceId : deviceIdList){ | ||
| for(String sensorId: sensorIdList){ | ||
| paths.add(new Path(deviceId + SEPARATOR_DEVIDE_SERIES + sensorId)); | ||
| } | ||
| } | ||
|
|
||
| if (timeFilter == null && freqFilter == null && valueFilter == null) { | ||
| return queryWithoutFilter(paths); | ||
| } else if (valueFilter instanceof SingleSeriesFilterExpression || (timeFilter != null && valueFilter == null)) { | ||
| return readOneColumnValueUseFilter(paths, (SingleSeriesFilterExpression) timeFilter, (SingleSeriesFilterExpression) freqFilter, | ||
| (SingleSeriesFilterExpression) valueFilter); | ||
| } else if (valueFilter instanceof CrossSeriesFilterExpression) { | ||
| return crossColumnQuery(paths, (SingleSeriesFilterExpression) timeFilter, (SingleSeriesFilterExpression) freqFilter, | ||
| (CrossSeriesFilterExpression) valueFilter); | ||
| } | ||
| throw new IOException("Query Not Support Exception"); | ||
| } | ||
|
|
||
| private QueryDataSet queryWithoutFilter(List<Path> paths) throws IOException { | ||
| return new IteratorQueryDataSet(paths) { | ||
| @Override | ||
| public DynamicOneColumnData getMoreRecordsForOneColumn(Path p, DynamicOneColumnData res) throws IOException { | ||
| return recordReader.getValueInOneColumnWithoutException(res, FETCH_SIZE, p.getDeltaObjectToString(), p.getMeasurementToString()); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| private QueryDataSet readOneColumnValueUseFilter(List<Path> paths, SingleSeriesFilterExpression timeFilter, | ||
| SingleSeriesFilterExpression freqFilter, SingleSeriesFilterExpression valueFilter) throws IOException { | ||
| logger.debug("start read one column data with filter"); | ||
|
|
||
| return new IteratorQueryDataSet(paths) { | ||
| @Override | ||
| public DynamicOneColumnData getMoreRecordsForOneColumn(Path p, DynamicOneColumnData res) throws IOException { | ||
| return recordReader.getValuesUseFilter(res, FETCH_SIZE, p.getDeltaObjectToString(), p.getMeasurementToString() | ||
| , timeFilter, freqFilter, valueFilter); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| private QueryDataSet crossColumnQuery(List<Path> paths, SingleSeriesFilterExpression timeFilter, SingleSeriesFilterExpression freqFilter, | ||
| CrossSeriesFilterExpression valueFilter) throws IOException { | ||
| CrossQueryTimeGenerator timeQueryDataSet = new CrossQueryTimeGenerator(timeFilter, freqFilter, valueFilter, FETCH_SIZE) { | ||
| @Override | ||
| public DynamicOneColumnData getDataInNextBatch(DynamicOneColumnData res, int fetchSize, | ||
| SingleSeriesFilterExpression valueFilter, int valueFilterNumber) throws ProcessorException, IOException { | ||
| return recordReader.getValuesUseFilter(res, fetchSize, valueFilter); | ||
| } | ||
| }; | ||
|
|
||
| return new CrossQueryIteratorDataSet(timeQueryDataSet) { | ||
| @Override | ||
| public boolean getMoreRecords() throws IOException { | ||
| try { | ||
| long[] timeRet = crossQueryTimeGenerator.generateTimes(); | ||
| if (timeRet.length == 0) { | ||
| return true; | ||
| } | ||
| for (Path p : paths) { | ||
| String deltaObjectUID = p.getDeltaObjectToString(); | ||
| String measurementUID = p.getMeasurementToString(); | ||
| DynamicOneColumnData oneColDataList = recordReader.getValuesUseTimestamps(deltaObjectUID, measurementUID, timeRet); | ||
| mapRet.put(p.getFullPath(), oneColDataList); | ||
| } | ||
|
|
||
| } catch (ProcessorException e) { | ||
| throw new IOException(e.getMessage()); | ||
| } | ||
| return false; | ||
| } | ||
| }; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.