[GitHub] carbondata pull request #2628: [CARBONDATA-2851][CARBONDATA-2852] Support zs...

[xuchuanyin]

Github user xuchuanyin commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/2628#discussion_r214561432
  
    --- Diff: 
core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java ---
    @@ -94,222 +98,240 @@ public void 
setStatsCollector(ColumnPageStatsCollector statsCollector) {
       }
     
       private static ColumnPage createDecimalPage(TableSpec.ColumnSpec 
columnSpec, DataType dataType,
    -      int pageSize) {
    +      int pageSize, String compressorName) {
         if (unsafe) {
           try {
    -        return new UnsafeDecimalColumnPage(columnSpec, dataType, pageSize);
    +        return new UnsafeDecimalColumnPage(columnSpec, dataType, pageSize, 
compressorName);
           } catch (MemoryException e) {
             throw new RuntimeException(e);
           }
         } else {
    -      return new SafeDecimalColumnPage(columnSpec, dataType, pageSize);
    +      return new SafeDecimalColumnPage(columnSpec, dataType, pageSize, 
compressorName);
         }
       }
     
       private static ColumnPage createVarLengthPage(TableSpec.ColumnSpec 
columnSpec, DataType dataType,
    -      int pageSize) {
    +      int pageSize, String compressorName) {
         if (unsafe) {
           try {
    -        return new UnsafeVarLengthColumnPage(columnSpec, dataType, 
pageSize);
    +        return new UnsafeVarLengthColumnPage(columnSpec, dataType, 
pageSize, compressorName);
           } catch (MemoryException e) {
             throw new RuntimeException(e);
           }
         } else {
    -      return new SafeVarLengthColumnPage(columnSpec, dataType, pageSize);
    +      return new SafeVarLengthColumnPage(columnSpec, dataType, pageSize, 
compressorName);
         }
       }
     
       private static ColumnPage createFixLengthPage(TableSpec.ColumnSpec 
columnSpec, DataType dataType,
    -      int pageSize) {
    +      int pageSize, String compressorName) {
         if (unsafe) {
           try {
    -        return new UnsafeFixLengthColumnPage(columnSpec, dataType, 
pageSize);
    +        return new UnsafeFixLengthColumnPage(columnSpec, dataType, 
pageSize, compressorName);
           } catch (MemoryException e) {
             throw new RuntimeException(e);
           }
         } else {
    -      return new SafeFixLengthColumnPage(columnSpec, dataType, pageSize);
    +      return new SafeFixLengthColumnPage(columnSpec, dataType, pageSize, 
compressorName);
         }
       }
     
       private static ColumnPage 
createFixLengthByteArrayPage(TableSpec.ColumnSpec columnSpec,
    -      DataType dataType, int pageSize, int eachValueSize) {
    +      DataType dataType, int pageSize, int eachValueSize, String 
compressorName) {
         if (unsafe) {
           try {
    -        return new UnsafeFixLengthColumnPage(columnSpec, dataType, 
pageSize, eachValueSize);
    +        return new UnsafeFixLengthColumnPage(columnSpec, dataType, 
pageSize, eachValueSize,
    +            compressorName);
           } catch (MemoryException e) {
             throw new RuntimeException(e);
           }
         } else {
    -      return new SafeFixLengthColumnPage(columnSpec, dataType, pageSize);
    +      return new SafeFixLengthColumnPage(columnSpec, dataType, pageSize, 
compressorName);
         }
       }
     
       private static ColumnPage createPage(TableSpec.ColumnSpec columnSpec, 
DataType dataType,
    -      int pageSize) {
    +      int pageSize, String compressorName) {
         if (DataTypes.isDecimal(dataType)) {
    -      return createDecimalPage(columnSpec, dataType, pageSize);
    +      return createDecimalPage(columnSpec, dataType, pageSize, 
compressorName);
         } else if (dataType.equals(BYTE_ARRAY)) {
    -      return createVarLengthPage(columnSpec, dataType, pageSize);
    +      return createVarLengthPage(columnSpec, dataType, pageSize, 
compressorName);
         } else {
    -      return createFixLengthPage(columnSpec, dataType, pageSize);
    +      return createFixLengthPage(columnSpec, dataType, pageSize, 
compressorName);
         }
       }
     
       public static ColumnPage newDecimalPage(TableSpec.ColumnSpec columnSpec, 
DataType dataType,
    -      int pageSize)
    +      int pageSize, String compressor)
         throws MemoryException {
    -    return newPage(columnSpec, dataType, pageSize);
    +    return newPage(columnSpec, dataType, pageSize, compressor);
       }
     
       public static ColumnPage newLocalDictPage(TableSpec.ColumnSpec 
columnSpec, DataType dataType,
           int pageSize, LocalDictionaryGenerator localDictionaryGenerator,
    -      boolean isComplexTypePrimitive) throws MemoryException {
    +      boolean isComplexTypePrimitive, String compressorName) throws 
MemoryException {
         if (unsafe) {
    -      return new LocalDictColumnPage(new 
UnsafeVarLengthColumnPage(columnSpec, dataType, pageSize),
    -          new UnsafeFixLengthColumnPage(columnSpec, DataTypes.BYTE_ARRAY, 
pageSize,
    -              CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE),
    -          localDictionaryGenerator, isComplexTypePrimitive);
    +      ColumnPage actualPage =
    +          new UnsafeVarLengthColumnPage(columnSpec, dataType, pageSize, 
compressorName);
    +      ColumnPage encodedPage = new UnsafeFixLengthColumnPage(columnSpec, 
DataTypes.BYTE_ARRAY,
    +          pageSize, 
CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE, compressorName);
    +      return new LocalDictColumnPage(actualPage, encodedPage, 
localDictionaryGenerator,
    +          isComplexTypePrimitive, compressorName);
         } else {
    -      return new LocalDictColumnPage(new 
SafeVarLengthColumnPage(columnSpec, dataType, pageSize),
    -          new SafeFixLengthColumnPage(columnSpec, DataTypes.BYTE_ARRAY, 
pageSize),
    -          localDictionaryGenerator, isComplexTypePrimitive);
    +      ColumnPage actualPage =
    +          new SafeVarLengthColumnPage(columnSpec, dataType, pageSize, 
compressorName);
    +      ColumnPage encodedPage =
    +          new SafeFixLengthColumnPage(columnSpec, DataTypes.BYTE_ARRAY, 
pageSize, compressorName);
    +      return new LocalDictColumnPage(actualPage, encodedPage, 
localDictionaryGenerator,
    --- End diff --
    
    ok, extract the code


---