Support default value read for ORC format in spark #76
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 hidden or 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 hidden or 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,120 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you 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 org.apache.iceberg.spark; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.stream.Collectors; | ||
| import org.apache.iceberg.MetadataColumns; | ||
| import org.apache.iceberg.orc.ORCSchemaUtil; | ||
| import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor; | ||
| import org.apache.iceberg.relocated.com.google.common.base.Preconditions; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Lists; | ||
| import org.apache.iceberg.spark.source.BaseDataReader; | ||
| import org.apache.iceberg.types.Types; | ||
| import org.apache.orc.TypeDescription; | ||
|
|
||
| public abstract class OrcSchemaWithTypeVisitorSpark<T> extends OrcSchemaWithTypeVisitor<T> { | ||
shenodaguirguis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| private final Map<Integer, Object> idToConstant; | ||
|
|
||
| public Map<Integer, Object> getIdToConstant() { | ||
| return idToConstant; | ||
| } | ||
|
|
||
| protected OrcSchemaWithTypeVisitorSpark(Map<Integer, ?> idToConstant) { | ||
| this.idToConstant = new HashMap<>(); | ||
| this.idToConstant.putAll(idToConstant); | ||
| } | ||
|
|
||
| @Override | ||
| protected T visitRecord( | ||
| Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) { | ||
| Preconditions.checkState( | ||
| icebergFiledIdsContainOrcFieldIdsInOrder(struct, record), | ||
| "Iceberg schema and ORC schema doesn't align, please call ORCSchemaUtil.buildOrcProjection" + | ||
| "to get an aligned ORC schema first!" | ||
| ); | ||
| List<Types.NestedField> iFields = struct.fields(); | ||
shenodaguirguis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| List<TypeDescription> fields = record.getChildren(); | ||
| List<String> names = record.getFieldNames(); | ||
| List<T> results = Lists.newArrayListWithExpectedSize(fields.size()); | ||
|
|
||
| for (int i = 0, j = 0; i < iFields.size(); i++) { | ||
| Types.NestedField iField = iFields.get(i); | ||
| TypeDescription field = j < fields.size() ? fields.get(j) : null; | ||
| if (field == null || (iField.fieldId() != ORCSchemaUtil.fieldId(field))) { | ||
| // there are 3 cases where we need to use idToConstant for an iField | ||
| // 1. The field is MetadataColumns.ROW_POSITION, we build a RowPositionReader | ||
| // 2. The field is a partition column, we build a ConstantReader | ||
| // 3. The field should be read using the default value, where we build a ConstantReader | ||
| // Here we should only need to update idToConstant when it's the 3rd case, | ||
| // because the first 2 cases have been handled by logic in PartitionUtil.constantsMap | ||
| if (!iField.equals(MetadataColumns.ROW_POSITION) && | ||
shenodaguirguis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| !idToConstant.containsKey(iField.fieldId())) { | ||
| idToConstant.put(iField.fieldId(), BaseDataReader.convertConstant(iField.type(), iField.getDefaultValue())); | ||
| } | ||
| } else { | ||
| results.add(visit(iField.type(), field, visitor)); | ||
| j++; | ||
| } | ||
| } | ||
| return visitor.record(struct, record, names, results); | ||
| } | ||
|
|
||
| private static boolean icebergFiledIdsContainOrcFieldIdsInOrder(Types.StructType struct, TypeDescription record) { | ||
| List<Integer> icebergIDList = struct.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toList()); | ||
| List<Integer> orcIDList = record.getChildren().stream().map(ORCSchemaUtil::fieldId).collect(Collectors.toList()); | ||
|
|
||
| return containsInOrder(icebergIDList, orcIDList); | ||
| } | ||
|
|
||
| /** | ||
| * Checks whether the first list contains all the integers | ||
| * in the same order as regarding to the second list, the first | ||
| * list can contain extra integers that the second list doesn't, | ||
| * but the ones that exist in the second list should occur in the | ||
| * same relative order in the first list. | ||
| * | ||
| * @param list1 the first list | ||
| * @param list2 the second list | ||
| * @return the condition check result | ||
| */ | ||
| private static boolean containsInOrder(List<Integer> list1, List<Integer> list2) { | ||
| if (list1.size() < list2.size()) { | ||
| return false; | ||
| } | ||
|
|
||
| for (int i = 0, j = 0; j < list2.size(); j++) { | ||
| if (i >= list1.size()) { | ||
|
There was a problem hiding this comment. hoist this check to the for loop condition? |
||
| return false; | ||
| } | ||
| while (!list1.get(i).equals(list2.get(j))) { | ||
| i++; | ||
| if (i >= list1.size()) { | ||
| return false; | ||
| } | ||
| } | ||
| i++; | ||
| } | ||
| return true; | ||
| } | ||
| } | ||
This file contains hidden or 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 hidden or 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,102 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you 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 org.apache.iceberg.spark.data.vectorized; | ||
|
|
||
| import org.apache.spark.sql.types.DataType; | ||
| import org.apache.spark.sql.types.Decimal; | ||
| import org.apache.spark.sql.vectorized.ColumnVector; | ||
| import org.apache.spark.sql.vectorized.ColumnarArray; | ||
| import org.apache.spark.sql.vectorized.ColumnarMap; | ||
| import org.apache.spark.unsafe.types.UTF8String; | ||
|
|
||
| public class ConstantArrayColumnVector extends ConstantColumnVector { | ||
|
|
||
| private final Object[] constantArray; | ||
|
|
||
| public ConstantArrayColumnVector(DataType type, int batchSize, Object[] constantArray) { | ||
| super(type, batchSize, constantArray); | ||
| this.constantArray = constantArray; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean getBoolean(int rowId) { | ||
| return (boolean) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public byte getByte(int rowId) { | ||
| return (byte) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public short getShort(int rowId) { | ||
| return (short) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public int getInt(int rowId) { | ||
| return (int) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public long getLong(int rowId) { | ||
| return (long) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public float getFloat(int rowId) { | ||
| return (float) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public double getDouble(int rowId) { | ||
| return (double) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public ColumnarArray getArray(int rowId) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public ColumnarMap getMap(int ordinal) { | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public Decimal getDecimal(int rowId, int precision, int scale) { | ||
| return (Decimal) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public UTF8String getUTF8String(int rowId) { | ||
| return (UTF8String) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| public byte[] getBinary(int rowId) { | ||
| return (byte[]) constantArray[rowId]; | ||
| } | ||
|
|
||
| @Override | ||
| protected ColumnVector getChild(int ordinal) { | ||
| return null; | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.