Fixed bug to parquet select columns. And add 'column count' variables

hoonki · hoonki · commit 32092a98b849 · 2020-07-28T15:27:39.000+09:00
diff --git a/common/data/src/main/java/com/samsung/sds/brightics/common/data/client/ParquetClient.java b/common/data/src/main/java/com/samsung/sds/brightics/common/data/client/ParquetClient.java
@@ -125,6 +125,7 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil
 				filteredColumns);
 		List<FileIndex> indexes = info.getLimitedFiles(min, max);
 		Column[] schema = info.getSchema();
+		int[] columnIndex = info.getColumnIndex();
 
 		List<Object[]> data = new ArrayList<>();
 		for (FileIndex index : indexes) {
@@ -145,13 +146,13 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil
 				}
 				long numRowCount = endIndex - startIndex - leftSkip - rightSkip;
 				reader = (BrighticsParquetReader<DefaultRecord>) BrighticsParquetUtils.getReader(new Path(filePath),
-						filteredColumns);
+						columnIndex);
 				while (leftSkip > 0) {
-					reader.read(filteredColumns);
+					reader.read(columnIndex);
 					leftSkip--;
 				}
 				DefaultRecord record;
-				while ((record = reader.read(filteredColumns)) != null && numRowCount > 0) {
+				while ((record = reader.read(columnIndex)) != null && numRowCount > 0) {
 					data.add(record.getValues());
 					numRowCount--;
 				}
@@ -163,7 +164,7 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil
 
 		}
 		//send filtered schema and row data information with total count and byte
-        return new ObjectTable(info.getCount(), info.getBytes(), schema, data);
+        return new ObjectTable(info.getCount(), schema.length, info.getBytes(), schema, data);
 	}
     
     public static ObjectTable readParquet(String path, long min, long max) throws IllegalArgumentException, IOException {
diff --git a/common/data/src/main/java/com/samsung/sds/brightics/common/data/parquet/reader/info/ParquetInformation.java b/common/data/src/main/java/com/samsung/sds/brightics/common/data/parquet/reader/info/ParquetInformation.java
@@ -1,39 +1,46 @@
-/*
-    Copyright 2019 Samsung SDS
-    
-    Licensed under the Apache License, Version 2.0 (the "License");
-    you may not use this file except in compliance with the License.
-    You may obtain a copy of the License at
-    
-        http://www.apache.org/licenses/LICENSE-2.0
-    
-    Unless required by applicable law or agreed to in writing, software
-    distributed under the License is distributed on an "AS IS" BASIS,
-    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-    See the License for the specific language governing permissions and
-    limitations under the License.
-*/
-
+/*
+    Copyright 2019 Samsung SDS
+    
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+    
+        http://www.apache.org/licenses/LICENSE-2.0
+    
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+*/
+
 package com.samsung.sds.brightics.common.data.parquet.reader.info;
 
 import java.util.List;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import com.samsung.sds.brightics.common.data.view.Column;
 
 public class ParquetInformation {
-    
+
     Column[] schema;
     long count;
     long bytes;
     List<FileIndex> page;
-    
+    int[] columnIndex;
+
     public ParquetInformation(Column[] schema, long count, long bytes, List<FileIndex> page) {
+        this(schema, count, bytes, page, new int[0]);
+    }
+
+    public ParquetInformation(Column[] schema, long count, long bytes, List<FileIndex> page, int[] columnIndex) {
         super();
         this.schema = schema;
         this.count = count;
         this.bytes = bytes;
         this.page = page;
+        this.columnIndex = columnIndex;
     }
 
     public List<FileIndex> getLimitedFiles(long lowerOffset, long upperOffset) {
@@ -46,6 +53,10 @@ public Column[] getSchema() {
         return schema;
     }
 
+    public int[] getColumnIndex() {
+        return columnIndex;
+    }
+
     public long getCount() {
         return count;
     }
@@ -57,7 +68,5 @@ public long getBytes() {
     public List<FileIndex> getPage() {
         return page;
     }
-    
-    
-    
+
 }
diff --git a/common/data/src/main/java/com/samsung/sds/brightics/common/data/parquet/reader/util/BrighticsParquetUtils.java b/common/data/src/main/java/com/samsung/sds/brightics/common/data/parquet/reader/util/BrighticsParquetUtils.java
@@ -16,15 +16,11 @@
 
 package com.samsung.sds.brightics.common.data.parquet.reader.util;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Iterator;
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
-
+import com.samsung.sds.brightics.common.data.parquet.reader.BrighticsParquetReadSupport;
+import com.samsung.sds.brightics.common.data.parquet.reader.DefaultRecord;
+import com.samsung.sds.brightics.common.data.parquet.reader.info.FileIndex;
+import com.samsung.sds.brightics.common.data.parquet.reader.info.ParquetInformation;
+import com.samsung.sds.brightics.common.data.view.Column;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -36,23 +32,23 @@
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.FileMetaData;
 import org.apache.parquet.hadoop.metadata.ParquetMetadata;
-import org.apache.parquet.schema.DecimalMetadata;
-import org.apache.parquet.schema.GroupType;
-import org.apache.parquet.schema.MessageType;
-import org.apache.parquet.schema.OriginalType;
-import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.*;
 import org.apache.parquet.schema.Type.Repetition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import com.samsung.sds.brightics.common.core.exception.BrighticsCoreException;
-import com.samsung.sds.brightics.common.data.parquet.reader.BrighticsParquetReadSupport;
-import com.samsung.sds.brightics.common.data.parquet.reader.DefaultRecord;
-import com.samsung.sds.brightics.common.data.parquet.reader.info.FileIndex;
-import com.samsung.sds.brightics.common.data.parquet.reader.info.ParquetInformation;
-import com.samsung.sds.brightics.common.data.view.Column;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 public class BrighticsParquetUtils {
 	
-//	private static final Logger LOGGER = LoggerFactory.getLogger("ParquetClient");
+	private static final Logger LOGGER = LoggerFactory.getLogger("ParquetClient");
 	
 	public static ParquetInformation getParquetInformation(Path path, Configuration conf, int[] filteredColumnIndex) throws IOException {
         FileStatus directory = FileSystem.get(conf).getFileStatus(path);
@@ -68,7 +64,8 @@ public static ParquetInformation getParquetInformation(Path path, Configuration
 		
         // set schema
         FileMetaData fileMetaData = footers.get(0).getParquetMetadata().getFileMetaData();
-        List<Type> filteredColumns = getFilteredColumns(fileMetaData.getSchema(), filteredColumnIndex);
+        int[] validatedColumnIndexArray = Arrays.stream(filteredColumnIndex).filter(i -> fileMetaData.getSchema().getColumns().size() > i).toArray();
+        List<Type> filteredColumns = getFilteredColumns(fileMetaData.getSchema(), validatedColumnIndexArray);
         Column[] schema = filteredColumns.stream().map(c -> new Column(c.getName(), convertTypeName(c))).toArray(Column[]::new);
         
         //set count, buffer size
@@ -86,7 +83,7 @@ public static ParquetInformation getParquetInformation(Path path, Configuration
             buf.add(new FileIndex(footer.getFile().toString(), previousTotalCount, previousTotalCount + currentCount));
             previousTotalCount += currentCount;
         }
-        return new ParquetInformation(schema, previousTotalCount, totalBytes, buf);
+        return new ParquetInformation(schema, previousTotalCount, totalBytes, buf, validatedColumnIndexArray);
     }
 	
     public static ParquetInformation getParquetInformation(Path path, Configuration conf) throws IOException {
@@ -194,7 +191,8 @@ public static List<Type> getFilteredColumns(MessageType schema, int[] filteredCo
         List<Type> copyFields = new ArrayList<Type>();
         //if filteredColumns is null or 0. pass filtering
         if(schema.getColumns().size() < filteredColumns.length) {
-        	throw new BrighticsCoreException("3102", "The column size used in the query is larger than the number of existing data columns.");
+            LOGGER.warn("The column size used in the query is larger than the number of existing data columns.");
+//        	throw new BrighticsCoreException("3102", "The column size used in the query is larger than the number of existing data columns.");
         }
         
         if (filteredColumns != null && filteredColumns.length > 0) {
@@ -220,7 +218,7 @@ public static int[] combineFilteredColumnIndexArray(int start, int end, int[] se
             Stream<Integer> range = IntStream.range(start, end + 1).boxed();
             return Stream.concat(range, selected).distinct().sorted().mapToInt(i -> i).toArray();
         } else {
-            return selected.mapToInt(i -> i).toArray();
+            return selected.distinct().sorted().mapToInt(i -> i).toArray();
         }
     }
 	
diff --git a/common/data/src/main/java/com/samsung/sds/brightics/common/data/view/ObjectTable.java b/common/data/src/main/java/com/samsung/sds/brightics/common/data/view/ObjectTable.java
@@ -1,19 +1,19 @@
-/*
-    Copyright 2019 Samsung SDS
-    
-    Licensed under the Apache License, Version 2.0 (the "License");
-    you may not use this file except in compliance with the License.
-    You may obtain a copy of the License at
-    
-        http://www.apache.org/licenses/LICENSE-2.0
-    
-    Unless required by applicable law or agreed to in writing, software
-    distributed under the License is distributed on an "AS IS" BASIS,
-    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-    See the License for the specific language governing permissions and
-    limitations under the License.
-*/
-
+/*
+    Copyright 2019 Samsung SDS
+    
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+    
+        http://www.apache.org/licenses/LICENSE-2.0
+    
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+*/
+
 package com.samsung.sds.brightics.common.data.view;
 
 import java.util.List;
@@ -32,4 +32,10 @@ public ObjectTable(long count, long bytes, Column[] schema, List<Object[]> data)
         this.data.put("data", data);
     }
 
+    public ObjectTable(long count, long columnCount, long bytes, Column[] schema, List<Object[]> data) {
+        super(count, bytes, schema);
+        this.data.put("columnCount", columnCount);
+        this.data.put("data", data);
+    }
+
 }

Original file line number	Diff line number	Diff line change
`@@ -125,6 +125,7 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil`
`125`	`125`	`filteredColumns);`
`126`	`126`	`List<FileIndex> indexes = info.getLimitedFiles(min, max);`
`127`	`127`	`Column[] schema = info.getSchema();`
	`128`	`+ int[] columnIndex = info.getColumnIndex();`
`128`	`129`
`129`	`130`	`List<Object[]> data = new ArrayList<>();`
`130`	`131`	`for (FileIndex index : indexes) {`
`@@ -145,13 +146,13 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil`
`145`	`146`	`}`
`146`	`147`	`long numRowCount = endIndex - startIndex - leftSkip - rightSkip;`
`147`	`148`	`reader = (BrighticsParquetReader<DefaultRecord>) BrighticsParquetUtils.getReader(new Path(filePath),`
`148`		`- filteredColumns);`
	`149`	`+ columnIndex);`
`149`	`150`	`while (leftSkip > 0) {`
`150`		`- reader.read(filteredColumns);`
	`151`	`+ reader.read(columnIndex);`
`151`	`152`	`leftSkip--;`
`152`	`153`	`}`
`153`	`154`	`DefaultRecord record;`
`154`		`- while ((record = reader.read(filteredColumns)) != null && numRowCount > 0) {`
	`155`	`+ while ((record = reader.read(columnIndex)) != null && numRowCount > 0) {`
`155`	`156`	`data.add(record.getValues());`
`156`	`157`	`numRowCount--;`
`157`	`158`	`}`
`@@ -163,7 +164,7 @@ public static ObjectTable readParquet(String path, long min, long max, int[] fil`
`163`	`164`
`164`	`165`	`}`
`165`	`166`	`//send filtered schema and row data information with total count and byte`
`166`		`- return new ObjectTable(info.getCount(), info.getBytes(), schema, data);`
	`167`	`+ return new ObjectTable(info.getCount(), schema.length, info.getBytes(), schema, data);`
`167`	`168`	`}`
`168`	`169`
`169`	`170`	`public static ObjectTable readParquet(String path, long min, long max) throws IllegalArgumentException, IOException {`