wypoon commented on code in PR #11661:
URL: https://github.com/apache/iceberg/pull/11661#discussion_r1859480431
##########
arrow/src/main/java/org/apache/iceberg/arrow/vectorized/parquet/VectorizedParquetDefinitionLevelReader.java:
##########
@@ -46,122 +47,217 @@ public VectorizedParquetDefinitionLevelReader(
super(bitWidth, maxDefLevel, readLength, setArrowValidityVector);
}
- abstract class NumericBaseReader {
- public void nextBatch(
+ abstract class CommonBaseReader {
+ protected void nextCommonBatch(
final FieldVector vector,
final int startOffset,
final int typeWidth,
final int numValsToRead,
NullabilityHolder nullabilityHolder,
- ValuesAsBytesReader valuesReader) {
- int bufferIdx = startOffset;
+ ValuesReader valuesReader,
+ Dictionary dict) {
+ int idx = startOffset;
int left = numValsToRead;
while (left > 0) {
if (currentCount == 0) {
readNextGroup();
}
int numValues = Math.min(left, currentCount);
+
+ byte[] byteArray = null;
+ if (typeWidth > -1) {
+ byteArray = new byte[typeWidth];
+ }
+ ArrowBuf validityBuffer = vector.getValidityBuffer();
+
switch (mode) {
case RLE:
- setNextNValuesInVector(
- typeWidth, nullabilityHolder, valuesReader, bufferIdx, vector,
numValues);
- bufferIdx += numValues;
+ if (valuesReader instanceof ValuesAsBytesReader) {
+ nextRleBatch(
+ vector,
+ typeWidth,
+ nullabilityHolder,
+ (ValuesAsBytesReader) valuesReader,
+ idx,
+ numValues,
+ byteArray);
+ } else if (valuesReader instanceof
VectorizedDictionaryEncodedParquetValuesReader) {
+ nextRleDictEncodedBatch(
+ vector,
+ typeWidth,
+ nullabilityHolder,
+ (VectorizedDictionaryEncodedParquetValuesReader)
valuesReader,
+ dict,
+ idx,
+ numValues,
+ validityBuffer);
+ }
+ idx += numValues;
break;
case PACKED:
- for (int i = 0; i < numValues; ++i) {
- if (packedValuesBuffer[packedValuesBufferIdx++] == maxDefLevel) {
- nextVal(vector, bufferIdx * typeWidth, valuesReader, mode);
- nullabilityHolder.setNotNull(bufferIdx);
- if (setArrowValidityVector) {
- BitVectorHelper.setBit(vector.getValidityBuffer(),
bufferIdx);
- }
- } else {
- setNull(nullabilityHolder, bufferIdx,
vector.getValidityBuffer());
- }
- bufferIdx++;
+ if (valuesReader instanceof ValuesAsBytesReader) {
+ nextPackedBatch(
+ vector,
+ typeWidth,
+ nullabilityHolder,
+ (ValuesAsBytesReader) valuesReader,
+ idx,
+ numValues,
+ byteArray);
+ } else if (valuesReader instanceof
VectorizedDictionaryEncodedParquetValuesReader) {
+ nextPackedDictEncodedBatch(
+ vector,
+ typeWidth,
+ nullabilityHolder,
+ (VectorizedDictionaryEncodedParquetValuesReader)
valuesReader,
+ dict,
+ idx,
+ numValues,
+ validityBuffer);
}
+ idx += numValues;
break;
}
left -= numValues;
currentCount -= numValues;
}
}
+ public void nextBatch(
+ final FieldVector vector,
+ final int startOffset,
+ final int typeWidth,
+ final int numValsToRead,
+ NullabilityHolder nullabilityHolder,
+ ValuesAsBytesReader valuesReader) {
+ nextCommonBatch(
+ vector, startOffset, typeWidth, numValsToRead, nullabilityHolder,
valuesReader, null);
+ }
+
public void nextDictEncodedBatch(
final FieldVector vector,
final int startOffset,
final int typeWidth,
final int numValsToRead,
NullabilityHolder nullabilityHolder,
- VectorizedDictionaryEncodedParquetValuesReader
dictionaryEncodedValuesReader,
+ VectorizedDictionaryEncodedParquetValuesReader valuesReader,
Dictionary dict) {
- int idx = startOffset;
- int left = numValsToRead;
- while (left > 0) {
- if (currentCount == 0) {
- readNextGroup();
- }
- int numValues = Math.min(left, currentCount);
- ArrowBuf validityBuffer = vector.getValidityBuffer();
- switch (mode) {
- case RLE:
- if (currentValue == maxDefLevel) {
- nextDictEncodedVal(
- vector,
- idx,
- dictionaryEncodedValuesReader,
- dict,
- mode,
- numValues,
- nullabilityHolder,
- typeWidth);
- } else {
- setNulls(nullabilityHolder, idx, numValues, validityBuffer);
- }
- idx += numValues;
- break;
- case PACKED:
- for (int i = 0; i < numValues; i++) {
- if (packedValuesBuffer[packedValuesBufferIdx++] == maxDefLevel) {
- nextDictEncodedVal(
- vector,
- idx,
- dictionaryEncodedValuesReader,
- dict,
- mode,
- numValues,
- nullabilityHolder,
- typeWidth);
- nullabilityHolder.setNotNull(idx);
- if (setArrowValidityVector) {
- BitVectorHelper.setBit(vector.getValidityBuffer(), idx);
- }
- } else {
- setNull(nullabilityHolder, idx, validityBuffer);
- }
- idx++;
- }
- break;
- }
- left -= numValues;
- currentCount -= numValues;
+ nextCommonBatch(
+ vector, startOffset, typeWidth, numValsToRead, nullabilityHolder,
valuesReader, dict);
+ }
+
+ protected abstract void nextRleBatch(
+ FieldVector vector,
+ int typeWidth,
+ NullabilityHolder nullabilityHolder,
+ ValuesAsBytesReader valuesReader,
+ int idx,
+ int numValues,
+ byte[] byteArray);
+
+ protected abstract void nextPackedBatch(
+ FieldVector vector,
+ int typeWidth,
+ NullabilityHolder nullabilityHolder,
+ ValuesAsBytesReader valuesReader,
+ int idx,
+ int numValues,
+ byte[] byteArray);
+
+ protected void nextRleDictEncodedBatch(
+ FieldVector vector,
+ int typeWidth,
+ NullabilityHolder nullabilityHolder,
+ VectorizedDictionaryEncodedParquetValuesReader valuesReader,
+ Dictionary dict,
+ int idx,
+ int numValues,
+ ArrowBuf validityBuffer) {
+ if (currentValue == maxDefLevel) {
+ nextDictEncodedVal(
+ vector, idx, valuesReader, dict, mode, numValues,
nullabilityHolder, typeWidth);
+ } else {
+ setNulls(nullabilityHolder, idx, numValues, validityBuffer);
}
}
- protected abstract void nextVal(
- FieldVector vector, int idx, ValuesAsBytesReader valuesReader, Mode
mode);
+ protected void nextPackedDictEncodedBatch(
+ FieldVector vector,
+ int typeWidth,
+ NullabilityHolder nullabilityHolder,
+ VectorizedDictionaryEncodedParquetValuesReader valuesReader,
+ Dictionary dict,
+ int idx,
+ int numValues,
+ ArrowBuf validityBuffer) {
+ int bufferIdx = idx;
Review Comment:
This is incremented in the method body and incrementing a parameter (`idx`)
is a checkstyle violation. Hence this variable.
The same applies to 3 other methods below.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
