Merge remote-tracking branch 'origin/modify-for-hadoop-connector' int…

…o modify-for-hadoop-connector
thulab · Dec 28, 2017 · 4643d07 · 4643d07
2 parents 182a08c + 305f3d8
commit 4643d07
Show file tree

Hide file tree

Showing 14 changed files with 162 additions and 103 deletions.
diff --git a/src/main/java/cn/edu/tsinghua/tsfile/common/conf/TSFileConfig.java b/src/main/java/cn/edu/tsinghua/tsfile/common/conf/TSFileConfig.java
@@ -55,7 +55,7 @@ public class TSFileConfig {
      */
     public int rleBitWidth = 8;
     public final int RLE_MIN_REPEATED_NUM = 8;
-    public final int RLE_MAX_REPEATED_NUM = 0x7FFFFF;
+    public final int RLE_MAX_REPEATED_NUM = 0x7FFF;
     public final int RLE_MAX_BIT_PACKED_NUM = 63;
 
     // Gorilla encoding configuration

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/IntPacker.java b/src/main/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/IntPacker.java
@@ -59,6 +59,7 @@ public void pack8Values(int[] values, int offset, byte[] buf) {
             if (leftBit > 0) {
                 buffer |= (values[valueIdx] << (32 - leftBit));
                 leftSize -= leftBit;
+                leftBit = 0;
                 valueIdx++;
             }
 

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/LongPacker.java b/src/main/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/LongPacker.java
@@ -60,6 +60,7 @@ public void pack8Values(long[] values, int offset, byte[] buf) {
             if (leftBit > 0) {
                 buffer |= (values[valueIdx] << (64 - leftBit));
                 leftSize -= leftBit;
+                leftBit = 0;
                 valueIdx++;
             }
 

diff --git a/...ilter/visitorImpl/IntervalTimeFilter.java → ...lter/visitorImpl/IntervalTimeVisitor.java b/...ilter/visitorImpl/IntervalTimeFilter.java → ...lter/visitorImpl/IntervalTimeVisitor.java
@@ -9,7 +9,7 @@
  *
  * @author CGF
  */
-public class IntervalTimeFilter implements FilterVisitor<Boolean> {
+public class IntervalTimeVisitor implements FilterVisitor<Boolean> {
 
     private Long startTime, endTime;
 

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/RecordReader.java b/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/RecordReader.java
@@ -191,15 +191,6 @@ public DynamicOneColumnData getValuesUseFilter(DynamicOneColumnData res, int fet
         return getValuesUseFilter(res, fetchSize, deltaObjectUID, measurementUID, null, null, valueFilter, idxs);
     }
 
-    private DynamicOneColumnData getValuesUseFilter(DynamicOneColumnData res, int fetchSize,
-                                                    RowGroupReader rowGroupReader, String measurementId, SingleSeriesFilterExpression timeFilter,
-                                                    SingleSeriesFilterExpression freqFilter, SingleSeriesFilterExpression valueFilter) throws IOException {
-
-        res = rowGroupReader.getValueReaders().get(measurementId).readOneColumnUseFilter(res, fetchSize, timeFilter,
-                freqFilter, valueFilter);
-        return res;
-    }
-
     public DynamicOneColumnData getValuesUseFilter(DynamicOneColumnData res, int fetchSize, String deltaObjectUID,
                                                    String measurementId, SingleSeriesFilterExpression timeFilter, SingleSeriesFilterExpression freqFilter,
                                                    SingleSeriesFilterExpression valueFilter, ArrayList<Integer> idxs) throws IOException {
@@ -231,6 +222,15 @@ public DynamicOneColumnData getValuesUseFilter(DynamicOneColumnData res, int fet
         return res;
     }
 
+    private DynamicOneColumnData getValuesUseFilter(DynamicOneColumnData res, int fetchSize,
+                                                    RowGroupReader rowGroupReader, String measurementId, SingleSeriesFilterExpression timeFilter,
+                                                    SingleSeriesFilterExpression freqFilter, SingleSeriesFilterExpression valueFilter) throws IOException {
+
+        res = rowGroupReader.getValueReaders().get(measurementId).readOneColumnUseFilter(res, fetchSize, timeFilter,
+                freqFilter, valueFilter);
+        return res;
+    }
+
     public DynamicOneColumnData getValuesUseTimestamps(String deltaObjectUID, String measurementId, long[] timestamps)
             throws IOException {
         checkSeries(deltaObjectUID, measurementId);

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/RowGroupReader.java b/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/RowGroupReader.java
@@ -54,7 +54,8 @@ public RowGroupReader(RowGroupMetaData rowGroupMetaData, ITsRandomAccessFileRead
                         tscMetaData.getVInTimeSeriesChunkMetaData().getDataType(),
                         tscMetaData.getVInTimeSeriesChunkMetaData().getDigest(), this.raf,
                         tscMetaData.getVInTimeSeriesChunkMetaData().getEnumValues(),
-                        tscMetaData.getProperties().getCompression(), tscMetaData.getNumRows());
+                        tscMetaData.getProperties().getCompression(), tscMetaData.getNumRows(),
+                        tscMetaData.getTInTimeSeriesChunkMetaData().getStartTime(), tscMetaData.getTInTimeSeriesChunkMetaData().getEndTime());
                 valueReaders.put(tscMetaData.getProperties().getMeasurementUID(), si);
             }
         }

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/ValueReader.java b/src/main/java/cn/edu/tsinghua/tsfile/timeseries/read/ValueReader.java
diff --git a/src/main/java/cn/edu/tsinghua/tsfile/timeseries/write/TsFileWriter.java b/src/main/java/cn/edu/tsinghua/tsfile/timeseries/write/TsFileWriter.java
@@ -1,5 +1,17 @@
 package cn.edu.tsinghua.tsfile.timeseries.write;
 
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.json.JSONObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import cn.edu.tsinghua.tsfile.common.conf.TSFileConfig;
 import cn.edu.tsinghua.tsfile.common.conf.TSFileDescriptor;
 import cn.edu.tsinghua.tsfile.common.utils.ITsRandomAccessFileWriter;
@@ -16,17 +28,6 @@
 import cn.edu.tsinghua.tsfile.timeseries.write.schema.converter.JsonConverter;
 import cn.edu.tsinghua.tsfile.timeseries.write.series.IRowGroupWriter;
 import cn.edu.tsinghua.tsfile.timeseries.write.series.RowGroupWriterImpl;
-import org.json.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
 
 /**
  * TsFileWriter is the entrance for writing processing. It receives a record and send it to

diff --git a/src/main/java/cn/edu/tsinghua/tsfile/timeseries/write/io/TsFileIOWriter.java b/src/main/java/cn/edu/tsinghua/tsfile/timeseries/write/io/TsFileIOWriter.java
@@ -247,10 +247,10 @@ private void serializeTsFileMetadata(TsFileMetaData footer) throws IOException {
      * @throws IOException if diff is greater than Integer.max_value
      */
     public void fillInRowGroup(long diff) throws IOException {
-        if (diff <= Integer.MAX_VALUE) {
-            out.write(new byte[(int) diff]);
+        if (diff <= 0) {
+            LOG.info("write too much blank byte array.");
         } else {
-            throw new IOException("write too much blank byte array!array size:" + diff);
+        	out.write(new byte[(int) diff]);
         }
     }
 

diff --git a/src/test/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/IntPackerTest.java b/src/test/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/IntPackerTest.java
@@ -44,4 +44,23 @@ public void test() {
 			assertEquals(res[i], v);
 		}
 	}
+
+	@Test
+	public void test2(){
+		for(int width = 4;width < 32; width++){
+			int[] arr = new int[8];
+			int[] res = new int[8];
+			for(int i = 0; i < 8; i++){
+				arr[i] = i;
+			}
+			IntPacker packer = new IntPacker(width);
+			byte[] buf = new byte[width];
+			packer.pack8Values(arr, 0, buf);
+			packer.unpack8Values(buf, 0, res);
+			for(int i = 0; i < 8; i++){
+				assertEquals(arr[i], res[i]);
+			}
+		}
+	}
+
 }
diff --git a/src/test/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/LongPackerTest.java b/src/test/java/cn/edu/tsinghua/tsfile/encoding/bitpacking/LongPackerTest.java
@@ -95,4 +95,22 @@ public void testPackAll() throws IOException {
 			assertEquals(readArray[i], v);
 		}
 	}
+
+	@Test
+	public void test2(){
+		for(int width = 4;width < 63; width++){
+			long[] arr = new long[8];
+			long[] res = new long[8];
+			for(int i = 0; i < 8; i++){
+				arr[i] = i;
+			}
+			LongPacker packer = new LongPacker(width);
+			byte[] buf = new byte[width];
+			packer.pack8Values(arr, 0, buf);
+			packer.unpack8Values(buf, 0, res);
+			for(int i = 0; i < 8; i++){
+				assertEquals(arr[i], res[i]);
+			}
+		}
+	}
 }
diff --git a/src/test/java/cn/edu/tsinghua/tsfile/encoding/decoder/IntRleDecoderTest.java b/src/test/java/cn/edu/tsinghua/tsfile/encoding/decoder/IntRleDecoderTest.java
@@ -78,14 +78,26 @@ public void setUp() throws Exception {
 			}
 			hybridCount += 2;
 		}
-
 		hybridWidth = ReadWriteStreamUtils.getIntMaxBitWidth(hybridList);
 	}
 
 	@After
 	public void tearDown() throws Exception {
 	}
 
+	@Test
+	public void testRleReadBigInt() throws IOException{
+		List<Integer> list = new ArrayList<>();
+		for(int i = 7000000; i < 10000000;i++){
+			list.add(i);
+		}
+		int width = ReadWriteStreamUtils.getIntMaxBitWidth(list);
+		testLength(list,width,false,1);
+		for(int i = 1;i < 10;i++){
+			testLength(list,width,false,i);
+		}
+	}
+
 	@Test
 	public void testRleReadInt() throws IOException{
 		for(int i = 1;i < 10;i++){

diff --git a/src/test/java/cn/edu/tsinghua/tsfile/encoding/decoder/LongRleDecoderTest.java b/src/test/java/cn/edu/tsinghua/tsfile/encoding/decoder/LongRleDecoderTest.java
@@ -87,6 +87,19 @@ public void setUp() throws Exception {
 	public void tearDown() throws Exception {
 	}
 
+	@Test
+	public void testRleReadBigLong() throws IOException{
+		List<Long> list = new ArrayList<>();
+		for(long i = 7000000; i < 10000000;i++){
+			list.add(i);
+		}
+		int width = ReadWriteStreamUtils.getLongMaxBitWidth(list);
+		testLength(list,width,false,1);
+		for(int i = 1;i < 10;i++){
+			testLength(list,width,false,i);
+		}
+	}
+
 	@Test
 	public void testRleReadLong() throws IOException{
 		for(int i = 1;i < 2;i++){

diff --git a/...series/filter/IntervalTimeFilterTest.java → ...eries/filter/IntervalTimeVisitorTest.java b/...series/filter/IntervalTimeFilterTest.java → ...eries/filter/IntervalTimeVisitorTest.java
@@ -8,7 +8,7 @@
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.And;
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.LtEq;
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.NotEq;
-import cn.edu.tsinghua.tsfile.timeseries.filter.visitorImpl.IntervalTimeFilter;
+import cn.edu.tsinghua.tsfile.timeseries.filter.visitorImpl.IntervalTimeVisitor;
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.GtEq;
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.Not;
 import cn.edu.tsinghua.tsfile.timeseries.filter.definition.operators.Or;
@@ -21,9 +21,9 @@
  * @author CGF
  *
  */
-public class IntervalTimeFilterTest {
+public class IntervalTimeVisitorTest {
 
-    private static final IntervalTimeFilter filter = new IntervalTimeFilter();
+    private static final IntervalTimeVisitor filter = new IntervalTimeVisitor();
 	private static String deltaObjectUID = FilterTestConstant.deltaObjectUID;
 	private static String measurementUID = FilterTestConstant.measurementUID;