You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/06/02 17:38:43 UTC
[GitHub] [spark] sunchao opened a new pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
sunchao opened a new pull request #32753:
URL: https://github.com/apache/spark/pull/32753
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660241499
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
I see, so there are 2 actions here:
1. get values from the parquet reader
2. put values into the spark columnar batch.
`skipBatch` does look confusing, we should use a better name, `skipValues`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868924203
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140345/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853359520
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun closed pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun closed pull request #32753:
URL: https://github.com/apache/spark/pull/32753
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660168778
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (currentStart == Long.MIN_VALUE) {
+ currentStart = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = idx;
+ }
+ previous = idx;
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
Review comment:
Should we assert `newOffset - offset` == `newRowId - rowId`?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
Review comment:
Rename it as `END_ROW_RANGE` or something else?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -151,26 +159,28 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- readState.resetForBatch(total);
+ readState.resetForNewBatch(total);
while (readState.valuesToReadInBatch > 0) {
- // Compute the number of values we want to read in this page.
if (readState.valuesToReadInPage == 0) {
int pageValueCount = readPage();
- readState.resetForPage(pageValueCount);
+ readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
// Save starting offset in case we need to decode dictionary IDs.
int startOffset = readState.offset;
+ // Save starting row index so we can check if we need to eagerly decode dict ids later
+ long startRowId = readState.rowId;
// Read and decode dictionary ids.
defColumn.readIntegers(readState, dictionaryIds, column,
(VectorizedValuesReader) dataColumn);
// TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process
// the values to add microseconds precision.
- if (column.hasDictionary() || (startOffset == 0 && isLazyDecodingSupported(typeName))) {
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
Review comment:
Hmm, based on the comment (`rowId != 0`) below, do we need to update it?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -295,6 +307,8 @@ private int readPageV1(DataPageV1 page) throws IOException {
}
private int readPageV2(DataPageV2 page) throws IOException {
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
Maybe move to `readPage`? Looks like `readPageV1` and `readPageV2` all need it.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
Review comment:
This is different to what I commented (https://github.com/apache/spark/pull/32753#discussion_r656582039) before. This looks more straightforward.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
I think it is because the name `skipBatch` somehow confuses. Actually it means to skip values from the reader.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
Maybe rename it? E.g. `skipFromReader`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868848755
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44876/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657284056
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -95,6 +80,18 @@
*/
private final ParquetVectorUpdaterFactory updaterFactory;
+ /**
+ * Helper struct to track intermediate states while reading Parquet pages in the column chunk.
+ */
+ private final ParquetReadState readState;
+
+ /**
+ * The index for the first row in the current page, among all rows across all pages in the
+ * column chunk for this reader. The value for this is 0 if there is no column index for the
Review comment:
Yes it can happen. Perhaps I should rephrase it to: "If there is no column index, the value for this is 0".
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657657426
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
##########
@@ -61,6 +61,14 @@ public final void readBooleans(int total, WritableColumnVector c, int rowId) {
}
}
+ @Override
+ public final void skipBooleans(int total) {
+ // TODO: properly vectorize this
Review comment:
Thanks!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853378395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659108572
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,107 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
offset = newOffset;
+ rowId = newRowId;
+ }
+
+ /**
+ * Advance to the next range.
+ */
+ void nextRange() {
+ if (rowRanges == null) {
+ currentRange = MAX_ROW_RANGE;
+ } else {
+ if (!rowRanges.hasNext()) {
+ currentRange = MIN_ROW_RANGE;
+ } else {
+ currentRange = rowRanges.next();
+ }
+ }
Review comment:
Shall we flatten more?
```java
if (rowRanges == null) {
currentRange = MAX_ROW_RANGE;
} else if (!rowRanges.hasNext()) {
currentRange = MIN_ROW_RANGE;
} else {
currentRange = rowRanges.next();
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661200729
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipValues(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
+ } else if (rowId > rangeEnd) {
+ state.nextRange();
+ } else {
+ // the range [rowId, rowId + n) overlaps with the current row range in state
+ long start = Math.max(rangeStart, rowId);
+ long end = Math.min(rangeEnd, rowId + n - 1);
+
+ // skip the part [rowId, start)
+ int toSkip = (int) (start - rowId);
+ if (toSkip > 0) {
Review comment:
To be safe, I'd like to recommend to move the type casting `(int)` into inside this `if` statement. For `if (toSkip > 0) {` check, we had better use `long`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658990194
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
No. The list of row ranges is associated with a Parquet row group. For example, let's say you have two columns `c1:int` and `c2:bigint`, and the following pages:
```
row index 0 500 1000 1500
-------------------------------
c1 (int) | | | |
-------------------------------
c2 (bigint) | | | | | | |
-------------------------------
0 250 500 750 1000 1250 1500
```
Suppose the query is `SELECT * FROM tbl WHERE c1 = 750 AND c2 = 1100`
This, when applied on `c1`, will produce row range `[500, 1000)`. When applied on `c2`, will produce row range `[1000, 1250)`. These two will be unioned into `[500, 1250)` and that is the row ranges for the whole row group.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867163244
@viirya I ran the benchmark added by @lxian in #31998 and here're the numbers:
```
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] simple filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 726 740 15 21.7 46.1 1.0X
[info] Parquet Vectorized (columnIndex) 30 34 4 533.1 1.9 24.6X
[info] Running benchmark: range filters
[info] Running case: Parquet Vectorized
[info] Stopped after 5 iterations, 4215 ms
[info] Running case: Parquet Vectorized (columnIndex)
[info] Stopped after 11 iterations, 2055 ms
[info] OpenJDK 64-Bit Server VM 1.8.0_282-b08 on Mac OS X 10.16
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] range filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 800 843 38 19.7 50.8 1.0X
[info] Parquet Vectorized (columnIndex) 165 187 34 95.4 10.5 4.9X
[info] Running benchmark: multi range filters
[info] Running case: Parquet Vectorized
[info] Stopped after 5 iterations, 4466 ms
[info] Running case: Parquet Vectorized (columnIndex)
[info] Stopped after 7 iterations, 2311 ms
[info] OpenJDK 64-Bit Server VM 1.8.0_282-b08 on Mac OS X 10.16
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] multi range filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 822 893 85 19.1 52.3 1.0X
[info] Parquet Vectorized (columnIndex) 305 330 16 51.5 19.4 2.7X
```
I also ran `DataSourceReadBenchmark` with and without the PR, and don't see much difference w.r.t Parquet vectorized read performance (although for some reason CSV read performance is quite different). The result is [here](https://gist.github.com/sunchao/e00947713dae790c6761ea860637d811).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655599949
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can compare indexes of values (rows) from a page with the row indexes mentioned above. This is provided by the `DataPage.getFirstRo
wIndex` method.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853378395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661193359
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +58,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
Review comment:
If we move this `rowIndexes == null ? null ...` part into `constructRanges` method, that will be better because it will protect both this constructor and `constructRanges` at the same time. For now, `constructRanges` seems to have the assumption that the argument is not `null` but there is no assertion for that. WDTY?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867002584
> Could we run parquet with/without column index benchmarks with this change?
Yes will do. Let me rebase the PR and run the benchmarks.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871010056
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140382/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867226319
**[Test build #140226 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140226/testReport)** for PR 32753 at commit [`7ccc510`](https://github.com/apache/spark/commit/7ccc5106fb0a3f2a710bb1edec7242d8209cf323).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659108216
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,107 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
Review comment:
If we always do `previous = idx` in all three cases, shall we simplify the logic by moving `previous = idx` out of the `if-else-statements`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853266303
**[Test build #139237 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139237/testReport)** for PR 32753 at commit [`89f90e9`](https://github.com/apache/spark/commit/89f90e9b29f362ee14fd5360141662049eb4bdd8).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-865208018
forgot to cc @wangyum and @rdblue
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655599949
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can keep consuming values from a page and skip them when they are not in the row indexes. This is provided by the `DataPage.getFirs
tRowIndex` method.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655599949
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can comparing indexes of values (rows) from a page with the row indexes. This is provided by the `DataPage.getFirstRowIndex` method
.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867312289
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140226/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659984532
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
It gives you an iterator so yeah generating them on the fly: https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/internal/filter2/columnindex/RowRanges.java#L253. The indexes are generated from `Range` which is very similar to what we defined here. I'm planning to file a JIRA in parquet-mr to just return the original `Range`s so we don't have to do this step in Spark.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867203052
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44750/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658990333
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,102 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
Review comment:
Yup will do
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853378395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853271761
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139237/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660177036
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (currentStart == Long.MIN_VALUE) {
+ currentStart = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = idx;
+ }
+ previous = idx;
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
Review comment:
I think this is not necessarily true: `rowId` tracks all the values that could be either read or skipped, while `offset` only tracks value that are read into the result column vector.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867312289
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140226/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655580221
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
I need more context to review this PR
1. where do we read the column index in `VectorizedColumnReader`?
2. where do we leverage the index to skip pages? In `VectorizedRleValuesReader`?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
If possible can we do this refactor first before adding column index support?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao edited a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao edited a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867163244
@viirya I ran the benchmark added by @lxian in #31998 and here're the numbers:
```
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] simple filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 726 740 15 21.7 46.1 1.0X
[info] Parquet Vectorized (columnIndex) 30 34 4 533.1 1.9 24.6X
[info] Running benchmark: range filters
[info] Running case: Parquet Vectorized
[info] Stopped after 5 iterations, 4215 ms
[info] Running case: Parquet Vectorized (columnIndex)
[info] Stopped after 11 iterations, 2055 ms
[info] OpenJDK 64-Bit Server VM 1.8.0_282-b08 on Mac OS X 10.16
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] range filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 800 843 38 19.7 50.8 1.0X
[info] Parquet Vectorized (columnIndex) 165 187 34 95.4 10.5 4.9X
[info] Running benchmark: multi range filters
[info] Running case: Parquet Vectorized
[info] Stopped after 5 iterations, 4466 ms
[info] Running case: Parquet Vectorized (columnIndex)
[info] Stopped after 7 iterations, 2311 ms
[info] OpenJDK 64-Bit Server VM 1.8.0_282-b08 on Mac OS X 10.16
[info] Intel(R) Core(TM) i9-10910 CPU @ 3.60GHz
[info] multi range filters: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
[info] ------------------------------------------------------------------------------------------------------------------------
[info] Parquet Vectorized 822 893 85 19.1 52.3 1.0X
[info] Parquet Vectorized (columnIndex) 305 330 16 51.5 19.4 2.7X
```
I also ran `DataSourceReadBenchmark` with and without the PR, and don't see much difference w.r.t Parquet vectorized read performance (although for some reason CSV read performance varied a lot between the two runs). The result is [here](https://gist.github.com/sunchao/e00947713dae790c6761ea860637d811).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853305123
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-870825440
**[Test build #140382 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140382/testReport)** for PR 32753 at commit [`8396b5f`](https://github.com/apache/spark/commit/8396b5fb03ecf606b0805fe99248ba6292007adf).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661200729
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipValues(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
+ } else if (rowId > rangeEnd) {
+ state.nextRange();
+ } else {
+ // the range [rowId, rowId + n) overlaps with the current row range in state
+ long start = Math.max(rangeStart, rowId);
+ long end = Math.min(rangeEnd, rowId + n - 1);
+
+ // skip the part [rowId, start)
+ int toSkip = (int) (start - rowId);
+ if (toSkip > 0) {
Review comment:
To be safe, I'd like to recommend to move the type casting `(int)` into inside this `if` statement. For `if (toSkip > 0) {` check, we had better use `long`. If the ranges are protected by line 191 ~ 192, then ignore this comment.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659964485
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
interesting, does the parquet read lib give you a big array containing these indexes, or it uses an algorithm to generate the indexes on the fly?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863562089
Thank you for updates, @sunchao !
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867311738
**[Test build #140226 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140226/testReport)** for PR 32753 at commit [`7ccc510`](https://github.com/apache/spark/commit/7ccc5106fb0a3f2a710bb1edec7242d8209cf323).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867241599
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44753/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660177273
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -151,26 +159,28 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- readState.resetForBatch(total);
+ readState.resetForNewBatch(total);
while (readState.valuesToReadInBatch > 0) {
- // Compute the number of values we want to read in this page.
if (readState.valuesToReadInPage == 0) {
int pageValueCount = readPage();
- readState.resetForPage(pageValueCount);
+ readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
// Save starting offset in case we need to decode dictionary IDs.
int startOffset = readState.offset;
+ // Save starting row index so we can check if we need to eagerly decode dict ids later
+ long startRowId = readState.rowId;
// Read and decode dictionary ids.
defColumn.readIntegers(readState, dictionaryIds, column,
(VectorizedValuesReader) dataColumn);
// TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process
// the values to add microseconds precision.
- if (column.hasDictionary() || (startOffset == 0 && isLazyDecodingSupported(typeName))) {
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
Review comment:
Oh yeah, I need to update the comment too
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655988235
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
+
+ /** The current row range */
+ private RowRange currentRange;
+
+ /** Maximum definition level */
+ int maxDefinitionLevel;
+
+ /** The current index overall all rows within the column chunk. This is used to check if the
+ * current row should be skipped by comparing the index against the row ranges. */
+ long rowId;
+
+ /** The offset to add the next value in the current batch */
Review comment:
add -> put? The first glance confused me. I read other method then know what this is for.
Or, "The offset in the current batch to put the next value"?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658990194
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
No. The list of row ranges is associated with a Parquet row group. For example, let's say you have two columns `c1:int` and `c2:bigint`, and the following pages:
```
row index 0 500 1000 1500
-------------------------------
c1 (int) | | | |
-------------------------------
c2 (bigint) | | | | | | |
-------------------------------
0 250 500 750 1000 1250 1500
```
Suppose the query is `SELECT * FROM tbl WHERE c1 = 750 AND c2 = 1100`
This, when applied on `c1`, will produce row ranges `[500, 1000)`. When applied on `c2`, will produce row ranges `[1000, 1250)`. These two will be unioned into `[500, 1250)` and that is the row ranges for the whole row group.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660179855
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
Review comment:
This is different to what I commented (https://github.com/apache/spark/pull/32753#discussion_r656582039) before. This looks more straightforward.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871674095
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44973/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660181125
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
I think it is because the name `skipBatch` somehow confuses. Actually it means to skip values from the reader.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868860007
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44876/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r644193462
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -171,479 +182,842 @@ public int readInteger() {
* }
*/
public void readIntegers(
- int total,
+ ParquetReadState state,
WritableColumnVector c,
- int rowId,
- int level,
VectorizedValuesReader data) throws IOException {
- int left = total;
- while (left > 0) {
+ int offset = state.offset;
+ long rowId = state.rowId;
+
+ while (state.hasMoreInPage(offset, rowId)) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
+ int n = Math.min(state.valuesToRead + state.offset - offset, this.currentCount);
switch (mode) {
case RLE:
- if (currentValue == level) {
- data.readIntegers(n, c, rowId);
+ if (!columnIndex.hasRowIndex()) {
+ if (currentValue == state.maxDefinitionLevel) {
+ data.readIntegers(n, c, offset);
+ } else {
+ c.putNulls(offset, n);
+ }
+ offset += n;
} else {
- c.putNulls(rowId, n);
+ if (currentValue == state.maxDefinitionLevel) {
Review comment:
We should be able to optimize this by checking if indexes within range `[rowId, rowId + n)` are all valid indexes, and if so call the batch version such as `data.readIntegers`. In fact it may be better to convert the `rowIndexes` into a sequence of row ranges just like how they were before.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853271716
**[Test build #139237 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139237/testReport)** for PR 32753 at commit [`89f90e9`](https://github.com/apache/spark/commit/89f90e9b29f362ee14fd5360141662049eb4bdd8).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `final class ParquetReadState `
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657484575
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -95,6 +80,18 @@
*/
private final ParquetVectorUpdaterFactory updaterFactory;
+ /**
+ * Helper struct to track intermediate states while reading Parquet pages in the column chunk.
+ */
+ private final ParquetReadState readState;
+
+ /**
+ * The index for the first row in the current page, among all rows across all pages in the
+ * column chunk for this reader. The value for this is 0 if there is no column index for the
Review comment:
yea, it sounds better.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655599949
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can keep consuming values from a page and skip them when they are not in the row indexes. This is provided by the `DataPage.getFirs
tRowIndex` method.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
Yes this occurred to me the other day as well :) I think it's a good idea. Let me move this refactoring part into a separate PR. Thanks.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can comparing indexes of values (rows) from a page with the row indexes. This is provided by the `DataPage.getFirstRowIndex` method
.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for complex schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can compare indexes of values (rows) from a page with the row indexes mentioned above. This is provided by the `DataPage.getFirstRo
wIndex` method.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for nested schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can compare indexes of values (rows) from a page with the row indexes mentioned above. This is provided by the `DataPage.getFirstRow
Index` method.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
Opened #33006
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868829589
**[Test build #140345 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140345/testReport)** for PR 32753 at commit [`4e16bbd`](https://github.com/apache/spark/commit/4e16bbd0948db19696eb296cf4189319e3adc05a).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659977651
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
No, because we are skipping the batch here, and not adding the values into the batch.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867244860
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44753/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659964485
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
interesting, does the parquet reader lib give you a big array containing these indexes, or it uses an algorithm to generate the indexes on the fly?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
And how fast/slow the parquet reader lib can generate the indexes?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660178570
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -295,6 +307,8 @@ private int readPageV1(DataPageV1 page) throws IOException {
}
private int readPageV2(DataPageV2 page) throws IOException {
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
Maybe move to `readPage`? Looks like `readPageV1` and `readPageV2` all need it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660169243
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
Review comment:
Rename it as `END_ROW_RANGE` or something else?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-870942258
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660357742
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
SGTM, I think `readValues` implies that it puts the values to the spark columnar batch
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853266303
**[Test build #139237 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/139237/testReport)** for PR 32753 at commit [`89f90e9`](https://github.com/apache/spark/commit/89f90e9b29f362ee14fd5360141662049eb4bdd8).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-870825440
**[Test build #140382 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140382/testReport)** for PR 32753 at commit [`8396b5f`](https://github.com/apache/spark/commit/8396b5fb03ecf606b0805fe99248ba6292007adf).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661195801
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdater.java
##########
@@ -30,20 +30,28 @@
* @param values destination values vector
* @param valuesReader reader to read values from
*/
- void updateBatch(
+ void readValues(
int total,
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader);
+ /**
+ * Skip a batch of `total` values from `valuesReader`.
+ *
+ * @param total total number of values to skip
+ * @param valuesReader reader to skip values from
+ */
+ void skipValues(int total, VectorizedValuesReader valuesReader);
Review comment:
Since this is renamed, please update the following PR description accordingly.
> introduced a new API ParquetVectorUpdater.skipBatch which skips a batch of values from a Parquet value reader.
And, maybe, we had better update line 40 in this file.
> Skip a batch of ...
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867179941
**[Test build #140222 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140222/testReport)** for PR 32753 at commit [`5c153da`](https://github.com/apache/spark/commit/5c153dad48fe20d9e068b20caebbee8a08189f6e).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659108805
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,107 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
offset = newOffset;
+ rowId = newRowId;
+ }
+
+ /**
+ * Advance to the next range.
+ */
+ void nextRange() {
+ if (rowRanges == null) {
+ currentRange = MAX_ROW_RANGE;
+ } else {
+ if (!rowRanges.hasNext()) {
+ currentRange = MIN_ROW_RANGE;
+ } else {
+ currentRange = rowRanges.next();
+ }
+ }
+ }
+
+ /**
+ * Helper struct to represent a range of row indexes `[start, end]`.
+ */
+ private static class RowRange {
+ long start;
Review comment:
If this should be immutable by definition of this struct, maybe `final`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659191689
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,107 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
Review comment:
Yeah good point. Will change.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659977651
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
No, because we are skipping the batch here, and not adding the values into the batch.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
It gives you an iterator so yeah generating them on the fly: https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/internal/filter2/columnindex/RowRanges.java#L253. The indexes are generated from `Range` which is very similar to what we defined here. I'm planning to file a JIRA in parquet-mr to just return the original `Range`s so we don't have to do this step in Spark.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
https://issues.apache.org/jira/browse/PARQUET-2061
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (currentStart == Long.MIN_VALUE) {
+ currentStart = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = idx;
+ }
+ previous = idx;
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
Review comment:
I think this is not necessarily true: `rowId` tracks all the values that could be either read or skipped, while `offset` only tracks value that are read into the result column vector.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -151,26 +159,28 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- readState.resetForBatch(total);
+ readState.resetForNewBatch(total);
while (readState.valuesToReadInBatch > 0) {
- // Compute the number of values we want to read in this page.
if (readState.valuesToReadInPage == 0) {
int pageValueCount = readPage();
- readState.resetForPage(pageValueCount);
+ readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
// Save starting offset in case we need to decode dictionary IDs.
int startOffset = readState.offset;
+ // Save starting row index so we can check if we need to eagerly decode dict ids later
+ long startRowId = readState.rowId;
// Read and decode dictionary ids.
defColumn.readIntegers(readState, dictionaryIds, column,
(VectorizedValuesReader) dataColumn);
// TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process
// the values to add microseconds precision.
- if (column.hasDictionary() || (startOffset == 0 && isLazyDecodingSupported(typeName))) {
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
Review comment:
Oh yeah, I need to update the comment too
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -295,6 +307,8 @@ private int readPageV1(DataPageV1 page) throws IOException {
}
private int readPageV2(DataPageV2 page) throws IOException {
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
Good point. Let me do that.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
> I see, so there are 2 actions here: ...
If you mean `readBatch`, then yes it does the two bullets above. For `skipBatch` it just skip the next `n` values from `valueReader`.
@viirya @cloud-fan Yeah I can change the method name to `skipValues` or `skipFromReader` if that's more clearer :) my preference, though, is to keep this close to the `readBatch` above. So how about I update the 3 methods in `ParquetVectorUpdater` to be: `readValues`, `readValue` and `skipValues`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871794031
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140459/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658923673
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
Review comment:
The `MIN_ROW_RANGE` here is only used as a place holder when we've iterated through all the ranges. It makes sure that we'll reject all row indexes that come after all the row ranges we've processed.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657290651
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
- int total,
- int offset,
+ ParquetReadState state,
WritableColumnVector values,
- int maxDefinitionLevel,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
- int left = total;
- while (left > 0) {
+ int offset = state.offset;
+ long rowId = state.rowId;
+
+ while (state.hasMoreInPage(offset, rowId)) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
- switch (mode) {
- case RLE:
- if (currentValue == maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(state.valuesToReadInBatch + state.offset - offset, this.currentCount);
Review comment:
because `state.valuesToReadInBatch` is the number of values left in the batch. Suppose the initial batch size is 1000, and `valuesToReadInBatch` is 400. This means we've read 600 values and so `state.offset` and `offset` are both 600 at the beginning. We'll need to use `offset - state.offset` to get how many values we've read so far.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867177138
**[Test build #140222 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140222/testReport)** for PR 32753 at commit [`5c153da`](https://github.com/apache/spark/commit/5c153dad48fe20d9e068b20caebbee8a08189f6e).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853271761
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/139237/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-870969054
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44905/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r653970103
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -340,6 +398,71 @@ public Binary readBinary(int len) {
throw new UnsupportedOperationException("only readInts is valid.");
}
+ @Override
+ public void skipIntegers(int total) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) this.readNextGroup();
+ int n = Math.min(left, this.currentCount);
+ advance(n);
+ left -= n;
+ }
+ }
+
+ @Override
+ public void skipBooleans(int total) {
+ throw new UnsupportedOperationException("only skipIntegers is supported");
Review comment:
Do we have a plan to support `skipBooleans`?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -340,6 +398,71 @@ public Binary readBinary(int len) {
throw new UnsupportedOperationException("only readInts is valid.");
}
+ @Override
+ public void skipIntegers(int total) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) this.readNextGroup();
+ int n = Math.min(left, this.currentCount);
+ advance(n);
+ left -= n;
+ }
+ }
+
+ @Override
+ public void skipBooleans(int total) {
+ throw new UnsupportedOperationException("only skipIntegers is supported");
Review comment:
`only skipIntegers is supported` has a little different meaning from `only skipIntegers is valid`. I'm curious what was the correct intention here.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
##########
@@ -378,11 +380,77 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSparkSession
.withWriterVersion(PARQUET_1_0)
.withCompressionCodec(GZIP)
.withRowGroupSize(1024 * 1024)
- .withPageSize(1024)
+ .withPageSize(pageSize)
+ .withDictionaryPageSize(dictionaryPageSize)
.withConf(hadoopConf)
.build()
}
+ test("test multiple pages with different sizes and nulls") {
Review comment:
For this test case, shall we add `SPARK-34859: ` prefix?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
##########
@@ -61,6 +61,14 @@ public final void readBooleans(int total, WritableColumnVector c, int rowId) {
}
}
+ @Override
+ public final void skipBooleans(int total) {
+ // TODO: properly vectorize this
Review comment:
Shall we file a JIRA and make this IDed TODO?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -340,6 +398,71 @@ public Binary readBinary(int len) {
throw new UnsupportedOperationException("only readInts is valid.");
}
+ @Override
+ public void skipIntegers(int total) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) this.readNextGroup();
+ int n = Math.min(left, this.currentCount);
+ advance(n);
+ left -= n;
+ }
+ }
+
+ @Override
+ public void skipBooleans(int total) {
+ throw new UnsupportedOperationException("only skipIntegers is supported");
Review comment:
+1 for changing.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661193359
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +58,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
Review comment:
If we move this `rowIndexes == null ? null ...` part into `constructRanges` method, that will be better because it will protect both this constructor and `constructRanges` at the same time. For now, `constructRanges` seems to have the assumption that the argument is not `null` but there is no assertion for that.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660190001
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
Maybe rename it? E.g. `skipFromReader`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661724331
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipValues(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
+ } else if (rowId > rangeEnd) {
+ state.nextRange();
+ } else {
+ // the range [rowId, rowId + n) overlaps with the current row range in state
+ long start = Math.max(rangeStart, rowId);
+ long end = Math.min(rangeEnd, rowId + n - 1);
+
+ // skip the part [rowId, start)
+ int toSkip = (int) (start - rowId);
+ if (toSkip > 0) {
Review comment:
`start` must >= `rowId` because it is defined as `long start = Math.max(rangeStart, rowId)`. Therefore, the case 1 `start < rowId` will never happen.
The second case, `(start - rowId) > Int.MaxValue`, can only occur if `start` is equal to `rangeStart`. In this case we also know that `rangeStart <= rowId + n` (from line 183) and `n` is `Math.min(leftInBatch, Math.min(leftInPage, this.currentCount))` which is guaranteed to be within integer range. Therefore, the cast is safe.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657715983
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,102 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart, previous;
+ currentStart = previous = Long.MIN_VALUE;
Review comment:
I'd prefer
```
long currentStart = Long.MIN_VALUE;
long previous = Long.MIN_VALUE;
```
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867179967
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140222/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661199457
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipValues(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
+ } else if (rowId > rangeEnd) {
+ state.nextRange();
+ } else {
+ // the range [rowId, rowId + n) overlaps with the current row range in state
+ long start = Math.max(rangeStart, rowId);
+ long end = Math.min(rangeEnd, rowId + n - 1);
+
+ // skip the part [rowId, start)
+ int toSkip = (int) (start - rowId);
+ if (toSkip > 0) {
Review comment:
Just a question. We may have two negative values.
1. start < rowId
2. (start - rowId) > Int.MaxValue
Are we considering both? Or, there is no change for case (2)?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipValues(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
+ } else if (rowId > rangeEnd) {
+ state.nextRange();
+ } else {
+ // the range [rowId, rowId + n) overlaps with the current row range in state
+ long start = Math.max(rangeStart, rowId);
+ long end = Math.min(rangeEnd, rowId + n - 1);
+
+ // skip the part [rowId, start)
+ int toSkip = (int) (start - rowId);
+ if (toSkip > 0) {
Review comment:
Just a question. We may have two negative value cases.
1. start < rowId
2. (start - rowId) > Int.MaxValue
Are we considering both? Or, there is no change for case (2)?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660234205
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -295,6 +307,8 @@ private int readPageV1(DataPageV1 page) throws IOException {
}
private int readPageV2(DataPageV2 page) throws IOException {
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
Good point. Let me do that.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867244841
Kubernetes integration test status failure
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44753/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871736077
Merged to master for Apache Spark 3.2.0. Thank you, @sunchao , @viirya , @cloud-fan
cc @aokolnychyi , @RussellSpitzer
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867244860
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44753/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658990194
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
No. The list of row ranges is associated with a Parquet row group. For example, let's say you have two columns `c1:int` and `c2:bigint`, and the following pages:
```
row index 0 500 1000 1500
-------------------------------
c1 (int) | | | |
-------------------------------
c2 (bigint) | | | | | | |
-------------------------------
0 250 500 750 1000 1250 1500
```
Suppose the query is `SELECT * FROM tbl WHERE c1 = 750 AND c2 = 1100`
This, when applied on `c1`, will produce row range `[500, 1000)`. When applied on `c2`, will produce row range `[1000, 1250)`. These two will be unioned into `[500, 1250)` and that is the row range for the whole row group.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868867598
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44876/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868829589
**[Test build #140345 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140345/testReport)** for PR 32753 at commit [`4e16bbd`](https://github.com/apache/spark/commit/4e16bbd0948db19696eb296cf4189319e3adc05a).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659962679
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
do we need to update `leftInBatch` here?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657471042
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
##########
@@ -61,6 +61,14 @@ public final void readBooleans(int total, WritableColumnVector c, int rowId) {
}
}
+ @Override
+ public final void skipBooleans(int total) {
+ // TODO: properly vectorize this
Review comment:
Created https://issues.apache.org/jira/browse/SPARK-35867
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657716760
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,102 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
Review comment:
Can we add some comments to explain how to generate row ranges?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871646642
**[Test build #140459 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140459/testReport)** for PR 32753 at commit [`6541d99`](https://github.com/apache/spark/commit/6541d99bd6788ae7ee3da1ccc76d30e9b39b9cb5).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660176278
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -151,26 +159,28 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- readState.resetForBatch(total);
+ readState.resetForNewBatch(total);
while (readState.valuesToReadInBatch > 0) {
- // Compute the number of values we want to read in this page.
if (readState.valuesToReadInPage == 0) {
int pageValueCount = readPage();
- readState.resetForPage(pageValueCount);
+ readState.resetForNewPage(pageValueCount, pageFirstRowIndex);
}
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
// Save starting offset in case we need to decode dictionary IDs.
int startOffset = readState.offset;
+ // Save starting row index so we can check if we need to eagerly decode dict ids later
+ long startRowId = readState.rowId;
// Read and decode dictionary ids.
defColumn.readIntegers(readState, dictionaryIds, column,
(VectorizedValuesReader) dataColumn);
// TIMESTAMP_MILLIS encoded as INT64 can't be lazily decoded as we need to post process
// the values to add microseconds precision.
- if (column.hasDictionary() || (startOffset == 0 && isLazyDecodingSupported(typeName))) {
+ if (column.hasDictionary() || (startRowId == pageFirstRowIndex &&
Review comment:
Hmm, based on the comment (`rowId != 0`) below, do we need to update it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-865208018
forgot to cc @wangyum and @rdblue
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r654010520
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -340,6 +398,71 @@ public Binary readBinary(int len) {
throw new UnsupportedOperationException("only readInts is valid.");
}
+ @Override
+ public void skipIntegers(int total) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) this.readNextGroup();
+ int n = Math.min(left, this.currentCount);
+ advance(n);
+ left -= n;
+ }
+ }
+
+ @Override
+ public void skipBooleans(int total) {
+ throw new UnsupportedOperationException("only skipIntegers is supported");
Review comment:
I can change it to `only skipIntegers is valid`. It's because Parquet RLE encoding only support integers which is why all the others are unimplemented.
##########
File path: sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetIOSuite.scala
##########
@@ -378,11 +380,77 @@ class ParquetIOSuite extends QueryTest with ParquetTest with SharedSparkSession
.withWriterVersion(PARQUET_1_0)
.withCompressionCodec(GZIP)
.withRowGroupSize(1024 * 1024)
- .withPageSize(1024)
+ .withPageSize(pageSize)
+ .withDictionaryPageSize(dictionaryPageSize)
.withConf(hadoopConf)
.build()
}
+ test("test multiple pages with different sizes and nulls") {
Review comment:
Sure wil do
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedPlainValuesReader.java
##########
@@ -61,6 +61,14 @@ public final void readBooleans(int total, WritableColumnVector c, int rowId) {
}
}
+ @Override
+ public final void skipBooleans(int total) {
+ // TODO: properly vectorize this
Review comment:
This follows a few other TODOs in the file when handling booleans. Yes I can file a JIRA too
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867203052
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44750/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661709891
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdater.java
##########
@@ -30,20 +30,28 @@
* @param values destination values vector
* @param valuesReader reader to read values from
*/
- void updateBatch(
+ void readValues(
int total,
int offset,
WritableColumnVector values,
VectorizedValuesReader valuesReader);
+ /**
+ * Skip a batch of `total` values from `valuesReader`.
+ *
+ * @param total total number of values to skip
+ * @param valuesReader reader to skip values from
+ */
+ void skipValues(int total, VectorizedValuesReader valuesReader);
Review comment:
Updated the PR description. Regarding the comment, IMO even though the method name is changed, the comment is still accurate in expressing what the method does.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r656479348
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
+
+ /** The current row range */
+ private RowRange currentRange;
+
+ /** Maximum definition level */
+ int maxDefinitionLevel;
+
+ /** The current index overall all rows within the column chunk. This is used to check if the
+ * current row should be skipped by comparing the index against the row ranges. */
+ long rowId;
+
+ /** The offset to add the next value in the current batch */
Review comment:
Thanks for the suggestion. I think that's more clear. Will change it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863496950
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868867598
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44876/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655600665
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
Yes this occurred to me the other day as well :) I think it's a good idea. Let me move this refactoring part into a separate PR. Thanks.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655599949
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
Sorry @cloud-fan, I should've add more context in the PR description. Let me try to add here and copy later to there.
1. The column index filtering is largely implemented in parquet-mr (via classes such as `ColumnIndex` and `ColumnIndexFilter`), and the filtered Parquet pages are returned to Spark through the `ParquetFileReader.readNextFilteredRowGroup` and `ParquetFileReader.getFilteredRecordCount` API. Please see #31393 for the related changes in the vectorized reader path.
2. Spark needs more work to handle mis-aligned Parquet pages returned from parquet-mr side, when there are multiple columns and their type width are different (e.g., int and bigint). For this issue, @lxian already gave a pretty good description in [SPARK-34859](https://issues.apache.org/jira/browse/SPARK-34859). To support the case, Spark needs to leverage the API [`PageReadStore.getRowIndexes`](https://javadoc.io/doc/org.apache.parquet/parquet-column/latest/org/apache/parquet/column/page/PageReadStore.html), which returns the indexes of all rows (note the difference between rows and values: for flat schema there is no difference between the two, but for nested schema they're different) after filtering within a Parquet row group. In addition, because there are gaps between pages, we'll need to know what is the index for the first row in a page, so we can compare indexes of values (rows) from a page with the row indexes mentioned above. This is provided by the `DataPage.getFirstRow
Index` method.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661706809
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +58,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
Review comment:
Yes that's a fair point. Will do.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871737030
Also, cc @gengliangwang since this was the release blocker for Apache Spark 3.2.0.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r656570701
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
+
+ /** The current row range */
+ private RowRange currentRange;
+
+ /** Maximum definition level */
+ int maxDefinitionLevel;
+
+ /** The current index overall all rows within the column chunk. This is used to check if the
+ * current row should be skipped by comparing the index against the row ranges. */
+ long rowId;
+
+ /** The offset to add the next value in the current batch */
+ int offset;
+
+ /** The remaining number of values to read in the current page */
+ int valuesToReadInPage;
+
+ /** The remaining number of values to read in the current batch */
+ int valuesToReadInBatch;
+
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart, previous;
+ currentStart = previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Called at the beginning of reading a new batch.
+ */
+ void resetForBatch(int batchSize) {
+ this.offset = 0;
+ this.valuesToReadInBatch = batchSize;
+ }
+
+ /**
+ * Called at the beginning of reading a new page.
+ */
+ void resetForPage(int totalValuesInPage, long pageFirstRowIndex) {
Review comment:
resetForNewPage?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
+
+ /** The current row range */
+ private RowRange currentRange;
+
+ /** Maximum definition level */
+ int maxDefinitionLevel;
+
+ /** The current index overall all rows within the column chunk. This is used to check if the
+ * current row should be skipped by comparing the index against the row ranges. */
+ long rowId;
+
+ /** The offset to add the next value in the current batch */
+ int offset;
+
+ /** The remaining number of values to read in the current page */
+ int valuesToReadInPage;
+
+ /** The remaining number of values to read in the current batch */
+ int valuesToReadInBatch;
+
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
+ this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart, previous;
+ currentStart = previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Called at the beginning of reading a new batch.
+ */
+ void resetForBatch(int batchSize) {
Review comment:
resetForNewBatch?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -174,24 +171,29 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- while (total > 0) {
+ readState.resetForBatch(total);
+ while (readState.valuesToReadInBatch > 0) {
// Compute the number of values we want to read in this page.
- int leftInPage = (int) (endOfPageValueCount - valuesRead);
- if (leftInPage == 0) {
+ if (readState.valuesToReadInPage == 0) {
readPage();
- leftInPage = (int) (endOfPageValueCount - valuesRead);
+ readState.resetForPage(pageValueCount, pageFirstRowIndex);
}
- int num = Math.min(total, leftInPage);
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
+ boolean supportLazyDecoding = readState.rowId == pageFirstRowIndex &&
Review comment:
Why `readState.rowId == pageFirstRowIndex` is a condition for supporting lazy decoding?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -174,24 +171,29 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- while (total > 0) {
+ readState.resetForBatch(total);
+ while (readState.valuesToReadInBatch > 0) {
// Compute the number of values we want to read in this page.
Review comment:
This comment is out-of-dated as `leftInPage` is removed.
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
- int total,
- int offset,
+ ParquetReadState state,
WritableColumnVector values,
- int maxDefinitionLevel,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
- int left = total;
- while (left > 0) {
+ int offset = state.offset;
+ long rowId = state.rowId;
+
+ while (state.hasMoreInPage(offset, rowId)) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
- switch (mode) {
- case RLE:
- if (currentValue == maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(state.valuesToReadInBatch + state.offset - offset, this.currentCount);
Review comment:
`n` is the number of values to read?
Isn't it just `Math.min(state.valuesToReadInBatch - offset, this.currentCount)`?
Why needs to add `state.offset`?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -95,6 +80,18 @@
*/
private final ParquetVectorUpdaterFactory updaterFactory;
+ /**
+ * Helper struct to track intermediate states while reading Parquet pages in the column chunk.
+ */
+ private final ParquetReadState readState;
+
+ /**
+ * The index for the first row in the current page, among all rows across all pages in the
+ * column chunk for this reader. The value for this is 0 if there is no column index for the
Review comment:
If there is column index, cannot `pageFirstRowIndex` be 0 if happened that it is exactly the first row to read?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -294,6 +292,7 @@ private void initDataReader(Encoding dataEncoding, ByteBufferInputStream in) thr
private void readPageV1(DataPageV1 page) throws IOException {
this.pageValueCount = page.getValueCount();
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
same question as above, cannot `getFirstRowIndex` return 0? Or the index begins with 1?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657714379
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
Review comment:
so `start` can be larger than `end` in `RowRange`?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660286736
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
> I see, so there are 2 actions here: ...
If you mean `readBatch`, then yes it does the two bullets above. For `skipBatch` it just skip the next `n` values from `valueReader`.
@viirya @cloud-fan Yeah I can change the method name to `skipValues` or `skipFromReader` if that's more clearer :) my preference, though, is to keep this close to the `readBatch` above. So how about I update the 3 methods in `ParquetVectorUpdater` to be: `readValues`, `readValue` and `skipValues`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655580942
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
If possible can we do this refactor first before adding column index support?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871784004
**[Test build #140459 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140459/testReport)** for PR 32753 at commit [`6541d99`](https://github.com/apache/spark/commit/6541d99bd6788ae7ee3da1ccc76d30e9b39b9cb5).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871693648
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44973/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-853378395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43760/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871646642
**[Test build #140459 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140459/testReport)** for PR 32753 at commit [`6541d99`](https://github.com/apache/spark/commit/6541d99bd6788ae7ee3da1ccc76d30e9b39b9cb5).
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863575428
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657285733
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -174,24 +171,29 @@ void readBatch(int total, WritableColumnVector column) throws IOException {
// page.
dictionaryIds = column.reserveDictionaryIds(total);
}
- while (total > 0) {
+ readState.resetForBatch(total);
+ while (readState.valuesToReadInBatch > 0) {
// Compute the number of values we want to read in this page.
- int leftInPage = (int) (endOfPageValueCount - valuesRead);
- if (leftInPage == 0) {
+ if (readState.valuesToReadInPage == 0) {
readPage();
- leftInPage = (int) (endOfPageValueCount - valuesRead);
+ readState.resetForPage(pageValueCount, pageFirstRowIndex);
}
- int num = Math.min(total, leftInPage);
PrimitiveType.PrimitiveTypeName typeName =
descriptor.getPrimitiveType().getPrimitiveTypeName();
if (isCurrentPageDictionaryEncoded) {
+ boolean supportLazyDecoding = readState.rowId == pageFirstRowIndex &&
Review comment:
This is just to check whether we are at the first iteration of the loop. Basically it checks whether lazy dictionary decoding is supported and if not, eagerly decode all the dictionary IDs read so far.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871710087
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44973/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863565038
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-870942258
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44905/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655580221
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -170,77 +170,135 @@ public int readInteger() {
* }
*/
public void readBatch(
Review comment:
I need more context to review this PR
1. where do we read the column index in `VectorizedColumnReader`?
2. where do we leverage the index to skip pages? In `VectorizedRleValuesReader`?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r655747755
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -0,0 +1,156 @@
+/*
+ * 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.spark.sql.execution.datasources.parquet;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
+/**
+ * Helper class to store intermediate state while reading a Parquet column chunk.
+ */
+final class ParquetReadState {
Review comment:
Opened #33006
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657286411
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
##########
@@ -294,6 +292,7 @@ private void initDataReader(Encoding dataEncoding, ByteBufferInputStream in) thr
private void readPageV1(DataPageV1 page) throws IOException {
this.pageValueCount = page.getValueCount();
+ this.pageFirstRowIndex = page.getFirstRowIndex().orElse(0L);
Review comment:
yea it can return 0, in which case it is the same as without column index and the whole page has to be read.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863565038
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657655917
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,102 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
+ }
+
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart, previous;
+ currentStart = previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (previous == Long.MIN_VALUE) {
+ currentStart = previous = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = previous = idx;
+ } else {
+ previous = idx;
+ }
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
}
/**
- * Called at the beginning of reading a new batch.
+ * Must be called at the beginning of reading a new batch.
*/
- void resetForBatch(int batchSize) {
+ void resetForNewBatch(int batchSize) {
Review comment:
Ya, new names (resetForNewBatch/resetForNewPage) look clear and better.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871004245
**[Test build #140382 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140382/testReport)** for PR 32753 at commit [`8396b5f`](https://github.com/apache/spark/commit/8396b5fb03ecf606b0805fe99248ba6292007adf).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867226319
**[Test build #140226 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140226/testReport)** for PR 32753 at commit [`7ccc510`](https://github.com/apache/spark/commit/7ccc5106fb0a3f2a710bb1edec7242d8209cf323).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r658990194
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
No. The list of row ranges is associated with a Parquet row group. For example, let's say you have two columns `c1:int` and `c2:bigint` in the row group, and the following pages:
```
row index 0 500 1000 1500
-------------------------------
c1 (int) | | | |
-------------------------------
c2 (bigint) | | | | | | |
-------------------------------
0 250 500 750 1000 1250 1500
```
Suppose the query is `SELECT * FROM tbl WHERE c1 = 750 AND c2 = 1100`
This, when applied on `c1`, will produce row range `[500, 1000)`. When applied on `c2`, will produce row range `[1000, 1250)`. These two will be unioned into `[500, 1250)` and that is the row range for the whole row group.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r657715329
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
does each column generate one row range?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659107932
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,31 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+ private static final RowRange MIN_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
Review comment:
Thank you for the illustration.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r661187202
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -17,13 +17,38 @@
package org.apache.spark.sql.execution.datasources.parquet;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.PrimitiveIterator;
+
/**
* Helper class to store intermediate state while reading a Parquet column chunk.
*/
final class ParquetReadState {
- /** Maximum definition level */
+ /** A special row range used when there is no row indexes (hence all rows must be included) */
+ private static final RowRange MAX_ROW_RANGE = new RowRange(Long.MIN_VALUE, Long.MAX_VALUE);
+
+ /**
+ * A special row range used when the row indexes are present AND all the row ranges have been
+ * processed. This serves as a sentinel at the end indicating that all rows come after the last
+ * row range should be skipped.
+ */
+ private static final RowRange END_ROW_RANGE = new RowRange(Long.MAX_VALUE, Long.MIN_VALUE);
+
+ /** Iterator over all row ranges, only not-null if column index is present */
+ private final Iterator<RowRange> rowRanges;
+
+ /** The current row range */
+ private RowRange currentRange;
+
+ /** Maximum definition level for the Parquet column */
final int maxDefinitionLevel;
+ /** The current index overall all rows within the column chunk. This is used to check if the
Review comment:
Maybe, `overall all` -> `over all`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867177138
**[Test build #140222 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140222/testReport)** for PR 32753 at commit [`5c153da`](https://github.com/apache/spark/commit/5c153dad48fe20d9e068b20caebbee8a08189f6e).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867179967
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140222/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659962679
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
do we need to update `leftInBatch` here?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
interesting, does the parquet read lib give you a big array containing these indexes, or it uses an algorithm to generate the indexes on the fly?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
interesting, does the parquet reader lib give you a big array containing these indexes, or it uses an algorithm to generate the indexes on the fly?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
And how fast/slow the parquet reader lib can generate the indexes?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
I see, so there are 2 actions here:
1. get values from the parquet reader
2. put values into the spark columnar batch.
`skipBatch` does look confusing, we should use a better name, `skipValues`?
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedRleValuesReader.java
##########
@@ -156,55 +156,81 @@ public int readInteger() {
}
/**
- * Reads `total` ints into `c` filling them in starting at `c[rowId]`. This reader
- * reads the definition levels and then will read from `data` for the non-null values.
- * If the value is null, c will be populated with `nullValue`. Note that `nullValue` is only
- * necessary for readIntegers because we also use it to decode dictionaryIds and want to make
- * sure it always has a value in range.
- *
- * This is a batched version of this logic:
- * if (this.readInt() == level) {
- * c[rowId] = data.readInteger();
- * } else {
- * c[rowId] = null;
- * }
+ * Reads a batch of values into vector `values`, using `valueReader`. The related states such
+ * as row index, offset, number of values left in the batch and page, etc, are tracked by
+ * `state`. The type-specific `updater` is used to update or skip values.
+ * <p>
+ * This reader reads the definition levels and then will read from `valueReader` for the
+ * non-null values. If the value is null, `values` will be populated with null value.
*/
public void readBatch(
ParquetReadState state,
WritableColumnVector values,
VectorizedValuesReader valueReader,
ParquetVectorUpdater updater) throws IOException {
int offset = state.offset;
- int left = Math.min(state.valuesToReadInBatch, state.valuesToReadInPage);
+ long rowId = state.rowId;
+ int leftInBatch = state.valuesToReadInBatch;
+ int leftInPage = state.valuesToReadInPage;
- while (left > 0) {
+ while (leftInBatch > 0 && leftInPage > 0) {
if (this.currentCount == 0) this.readNextGroup();
- int n = Math.min(left, this.currentCount);
-
- switch (mode) {
- case RLE:
- if (currentValue == state.maxDefinitionLevel) {
- updater.updateBatch(n, offset, values, valueReader);
- } else {
- values.putNulls(offset, n);
- }
- break;
- case PACKED:
- for (int i = 0; i < n; ++i) {
- if (currentBuffer[currentBufferIdx++] == state.maxDefinitionLevel) {
- updater.update(offset + i, values, valueReader);
+ int n = Math.min(leftInBatch, Math.min(leftInPage, this.currentCount));
+
+ long rangeStart = state.currentRangeStart();
+ long rangeEnd = state.currentRangeEnd();
+
+ if (rowId + n < rangeStart) {
+ updater.skipBatch(n, valueReader);
+ advance(n);
+ rowId += n;
+ leftInPage -= n;
Review comment:
SGTM, I think `readValues` implies that it puts the values to the spark columnar batch
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on pull request #32753: [WIP][SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863561674
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] sunchao commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
sunchao commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r659989472
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
Review comment:
https://issues.apache.org/jira/browse/PARQUET-2061
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32753:
URL: https://github.com/apache/spark/pull/32753#discussion_r660168778
##########
File path: sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetReadState.java
##########
@@ -33,31 +51,104 @@
/** The remaining number of values to read in the current batch */
int valuesToReadInBatch;
- ParquetReadState(int maxDefinitionLevel) {
+ ParquetReadState(int maxDefinitionLevel, PrimitiveIterator.OfLong rowIndexes) {
this.maxDefinitionLevel = maxDefinitionLevel;
+ this.rowRanges = rowIndexes == null ? null : constructRanges(rowIndexes);
+ nextRange();
}
/**
- * Called at the beginning of reading a new batch.
+ * Construct a list of row ranges from the given `rowIndexes`. For example, suppose the
+ * `rowIndexes` are `[0, 1, 2, 4, 5, 7, 8, 9]`, it will be converted into 3 row ranges:
+ * `[0-2], [4-5], [7-9]`.
*/
- void resetForBatch(int batchSize) {
+ private Iterator<RowRange> constructRanges(PrimitiveIterator.OfLong rowIndexes) {
+ List<RowRange> rowRanges = new ArrayList<>();
+ long currentStart = Long.MIN_VALUE;
+ long previous = Long.MIN_VALUE;
+
+ while (rowIndexes.hasNext()) {
+ long idx = rowIndexes.nextLong();
+ if (currentStart == Long.MIN_VALUE) {
+ currentStart = idx;
+ } else if (previous + 1 != idx) {
+ RowRange range = new RowRange(currentStart, previous);
+ rowRanges.add(range);
+ currentStart = idx;
+ }
+ previous = idx;
+ }
+
+ if (previous != Long.MIN_VALUE) {
+ rowRanges.add(new RowRange(currentStart, previous));
+ }
+
+ return rowRanges.iterator();
+ }
+
+ /**
+ * Must be called at the beginning of reading a new batch.
+ */
+ void resetForNewBatch(int batchSize) {
this.offset = 0;
this.valuesToReadInBatch = batchSize;
}
/**
- * Called at the beginning of reading a new page.
+ * Must be called at the beginning of reading a new page.
*/
- void resetForPage(int totalValuesInPage) {
+ void resetForNewPage(int totalValuesInPage, long pageFirstRowIndex) {
this.valuesToReadInPage = totalValuesInPage;
+ this.rowId = pageFirstRowIndex;
}
/**
- * Advance the current offset to the new values.
+ * Returns the start index of the current row range.
*/
- void advanceOffset(int newOffset) {
+ long currentRangeStart() {
+ return currentRange.start;
+ }
+
+ /**
+ * Returns the end index of the current row range.
+ */
+ long currentRangeEnd() {
+ return currentRange.end;
+ }
+
+ /**
+ * Advance the current offset and rowId to the new values.
+ */
+ void advanceOffsetAndRowId(int newOffset, long newRowId) {
valuesToReadInBatch -= (newOffset - offset);
- valuesToReadInPage -= (newOffset - offset);
+ valuesToReadInPage -= (newRowId - rowId);
Review comment:
Should we assert `newOffset - offset` == `newRowId - rowId`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-867196485
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44750/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868901657
**[Test build #140345 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/140345/testReport)** for PR 32753 at commit [`4e16bbd`](https://github.com/apache/spark/commit/4e16bbd0948db19696eb296cf4189319e3adc05a).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871794031
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140459/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-871710087
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/44973/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-863750006
sorry, I'm not very familiar with the parquet column index. does it allow us to skip pages?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32753: [SPARK-34859][SQL] Handle column index when using vectorized Parquet reader
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32753:
URL: https://github.com/apache/spark/pull/32753#issuecomment-868924203
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/140345/
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org