You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/06/03 16:53:33 UTC
[GitHub] [hudi] alexeykudinkin opened a new pull request, #5746: [WIP] Optimize performance of Column Stats filtering seq
alexeykudinkin opened a new pull request, #5746:
URL: https://github.com/apache/hudi/pull/5746
## *Tips*
- *Thank you very much for contributing to Apache Hudi.*
- *Please review https://hudi.apache.org/contribute/how-to-contribute before opening a pull request.*
## What is the purpose of the pull request
Currently from our benchmarking, Column Stats has somewhat static overhead of ~1-2s / table (~70000 files).
This PR is taking a stab at eliminating this overhead:
- Avoiding capturing heavy objets in closure requiring extensive cleaning by Spark
- ...
## Brief change log
- TBD
## Verify this pull request
This pull request is already covered by existing tests, such as *(please describe tests)*.
## Committer checklist
- [ ] Has a corresponding JIRA in PR title & commit
- [ ] Commit message is descriptive of the change
- [ ] CI is green
- [ ] Necessary doc changes done or have another open PR
- [ ] For large changes, please consider breaking it into sub-tasks under an umbrella JIRA.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929283416
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCatalystUtils.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hudi
+
+import org.apache.hudi.common.data.HoodieData
+import org.apache.spark.sql.Dataset
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.storage.StorageLevel._
+
+object HoodieCatalystUtils extends SparkAdapterSupport {
+
+ /**
+ * Executes provided function while keeping provided [[Dataset]] instance persisted for the
+ * duration of the execution
+ *
+ * @param df target [[Dataset]] to be persisted
+ * @param level desired [[StorageLevel]] of the persistence
+ * @param f target function to be executed while [[Dataset]] is kept persisted
+ * @tparam T return value of the target function
+ * @return execution outcome of the [[f]] function
+ */
+ def withPersistedDataset[T](df: Dataset[_], level: StorageLevel = MEMORY_AND_DISK)(f: => T): T = {
+ df.persist(level)
+ try {
+ f
+ } finally {
+ df.unpersist()
+ }
Review Comment:
Correct
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1194670776
CI is green:
<img width="1534" alt="Screen Shot 2022-07-25 at 2 45 06 PM" src="https://user-images.githubusercontent.com/428277/180879241-3249b2f4-def2-4b2d-91c0-5e0a59f2aa18.png">
https://dev.azure.com/apache-hudi-ci-org/apache-hudi-ci/_build/results?buildId=10321&view=results
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] YannByron commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
YannByron commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928087048
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java:
##########
@@ -170,7 +170,8 @@ Map<Pair<String, String>, HoodieMetadataColumnStats> getColumnStats(final List<P
* @return {@link HoodieData} of {@link HoodieRecord}s with records matching the passed in key prefixes.
*/
HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(List<String> keyPrefixes,
- String partitionName);
+ String partitionName,
+ boolean shouldLoadInMemory);
Review Comment:
Whether or not column statistics should be loaded in memory is the behavior of a metadata table.
IMO, we can abstract a method named `shouldLoadColumnStatsInMemory` in `HoodieTableMetadata`, the sub-classes can implement this, and use this in their own `getRecordsByKeyPrefixes` directly.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1191943032
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115)
* 0a15f230f412c1670dbf32465e99fe47af77c066 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193038327
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 909fb5e89b8e9e2e84a6f4892025ddb3715497f5 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1191985735
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0a15f230f412c1670dbf32465e99fe47af77c066 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1190924492
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
* d3a78d17d2c9b747df8673042f08c218c6d0b331 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113)
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929039025
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,18 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK = "spark";
Review Comment:
Good call. This shouldn't be Spark this should an engine-neutral
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1190887713
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
* d3a78d17d2c9b747df8673042f08c218c6d0b331 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929200613
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,26 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE = "engine";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
+ .noDefaultValue()
+ .withValidValues(COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY, COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE)
+ .sinceVersion("0.12.0")
+ .withDocumentation("By default Column Stats Index is automatically determining whether it should be read and processed either"
+ + "'in-memory' (w/in executing process) or using Spark (on a cluster), based on some factors like the size of the Index "
Review Comment:
Don't forget to update the docs as well
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +199,184 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
Review Comment:
This works for scala 2.12 as well?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCatalystUtils.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hudi
+
+import org.apache.hudi.common.data.HoodieData
+import org.apache.spark.sql.Dataset
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.storage.StorageLevel._
+
+object HoodieCatalystUtils extends SparkAdapterSupport {
+
+ /**
+ * Executes provided function while keeping provided [[Dataset]] instance persisted for the
+ * duration of the execution
+ *
+ * @param df target [[Dataset]] to be persisted
+ * @param level desired [[StorageLevel]] of the persistence
+ * @param f target function to be executed while [[Dataset]] is kept persisted
+ * @tparam T return value of the target function
+ * @return execution outcome of the [[f]] function
+ */
+ def withPersistedDataset[T](df: Dataset[_], level: StorageLevel = MEMORY_AND_DISK)(f: => T): T = {
+ df.persist(level)
+ try {
+ f
+ } finally {
+ df.unpersist()
+ }
Review Comment:
I remember you put similar util methods in before. Are these the same?
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,26 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE = "engine";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
+ .noDefaultValue()
+ .withValidValues(COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY, COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE)
+ .sinceVersion("0.12.0")
+ .withDocumentation("By default Column Stats Index is automatically determining whether it should be read and processed either"
+ + "'in-memory' (w/in executing process) or using Spark (on a cluster), based on some factors like the size of the Index "
+ + "and how many columns are read. This config allows to override this behavior.");
+
+ public static final ConfigProperty<Integer> COLUMN_STATS_INDEX_IN_MEMORY_PROJECTION_THRESHOLD = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.inMemory.projection.threshold")
+ .defaultValue(100000)
Review Comment:
Any perf numbers to support this default threshold?
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +199,184 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
Review Comment:
Basically, this is still spark-specific. Later on, we need to think about how to generalize this to non-Spark environment.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1194370120
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
}, {
"hash" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10321",
"triggerID" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a7ba7493c4f35b74082d8d3bf66246feaf9a91a6 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248)
* b1e993b6f13c44a532a21470d0519329e2cd42be Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10321)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1191021230
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] XuQianJin-Stars commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
XuQianJin-Stars commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928182362
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
+ val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePathV2.toString)
// Read Metadata Table's Column Stats Index into Spark's [[DataFrame]]
- spark.read.format("org.apache.hudi")
+ val colStatsDF = spark.read.format("org.apache.hudi")
.options(metadataConfig.getProps.asScala)
.load(s"$metadataTablePath/${MetadataPartitionType.COLUMN_STATS.getPartitionPath}")
- }
-
- private def readColumnStatsIndexForColumnsInternal(spark: SparkSession, targetColumns: Seq[String], metadataConfig: HoodieMetadataConfig, tableBasePath: String) = {
- val ctx = new HoodieSparkEngineContext(new JavaSparkContext(spark.sparkContext))
- // Read Metadata Table's Column Stats Index into Spark's [[DataFrame]] by
- // - Fetching the records from CSI by key-prefixes (encoded column names)
- // - Deserializing fetched records into [[InternalRow]]s
- // - Composing [[DataFrame]]
- val metadataTableDF = {
- val metadataTable = HoodieTableMetadata.create(ctx, metadataConfig, tableBasePath, FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue)
-
- // TODO encoding should be done internally w/in HoodieBackedTableMetadata
- val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
-
- val recordsRDD: RDD[HoodieRecord[HoodieMetadataPayload]] =
- HoodieJavaRDD.getJavaRDD(
- metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)
- )
-
- val catalystRowsRDD: RDD[InternalRow] = recordsRDD.mapPartitions { it =>
- val metadataRecordSchema = new Parser().parse(metadataRecordSchemaString)
- val converter = AvroConversionUtils.createAvroToInternalRowConverter(metadataRecordSchema, metadataRecordStructType)
-
- it.map { record =>
- // schema and props are ignored for generating metadata record from the payload
- // instead, the underlying file system, or bloom filter, or columns stats metadata (part of payload) are directly used
- toScalaOption(record.getData.getInsertValue(null, null))
- .flatMap(avroRecord => converter(avroRecord.asInstanceOf[GenericRecord]))
- .orNull
- }
- }
+ val requiredIndexColumns =
+ targetColumnStatsIndexColumns.map(colName =>
+ col(s"${HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS}.${colName}"))
- HoodieUnsafeRDDUtils.createDataFrame(spark, catalystRowsRDD, metadataRecordStructType)
- }
- metadataTableDF
+ colStatsDF.where(col(HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS).isNotNull)
+ .select(requiredIndexColumns: _*)
}
}
object ColumnStatsIndexSupport {
- private val metadataRecordSchemaString: String = HoodieMetadataRecord.SCHEMA$.toString
- private val metadataRecordStructType: StructType = AvroConversionUtils.convertAvroSchemaToStructType(HoodieMetadataRecord.SCHEMA$)
+ private val columnStatsIndexProjectionSizeInMemoryThreshold = 100000
Review Comment:
> could you explain a little why set `columnStatsIndexProjectionSizeInMemoryThreshold` to 100,000?
Is it possible to configure by parameter default 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.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1191939594
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115)
* 0a15f230f412c1670dbf32465e99fe47af77c066 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193014296
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0a15f230f412c1670dbf32465e99fe47af77c066 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164)
* 909fb5e89b8e9e2e84a6f4892025ddb3715497f5 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928972329
##########
hudi-common/src/main/java/org/apache/hudi/BaseHoodieTableFileIndex.java:
##########
@@ -166,6 +167,11 @@ public Map<String, List<FileSlice>> listFileSlices() {
.collect(Collectors.toMap(e -> e.getKey().path, Map.Entry::getValue));
}
+ public int getFileSlicesCount() {
+ return cachedAllInputFileSlices.values().stream()
+ .reduce(0, (count, fileSlices) -> count + fileSlices.size(), Integer::sum);
Review Comment:
This is not quite obvious. Do sth like `cachedAllInputFileSlices.values().stream().mapToInt(List::size).sum()` instead?
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,18 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK = "spark";
Review Comment:
nit: this can be enums.
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,18 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK = "spark";
Review Comment:
`COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK`: based on the logic, this mode is more of leveraging engine context, could be Spark or Flink (plain Java), right?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] YannByron commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
YannByron commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928089469
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
+ val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePathV2.toString)
// Read Metadata Table's Column Stats Index into Spark's [[DataFrame]]
- spark.read.format("org.apache.hudi")
+ val colStatsDF = spark.read.format("org.apache.hudi")
.options(metadataConfig.getProps.asScala)
.load(s"$metadataTablePath/${MetadataPartitionType.COLUMN_STATS.getPartitionPath}")
- }
-
- private def readColumnStatsIndexForColumnsInternal(spark: SparkSession, targetColumns: Seq[String], metadataConfig: HoodieMetadataConfig, tableBasePath: String) = {
- val ctx = new HoodieSparkEngineContext(new JavaSparkContext(spark.sparkContext))
- // Read Metadata Table's Column Stats Index into Spark's [[DataFrame]] by
- // - Fetching the records from CSI by key-prefixes (encoded column names)
- // - Deserializing fetched records into [[InternalRow]]s
- // - Composing [[DataFrame]]
- val metadataTableDF = {
- val metadataTable = HoodieTableMetadata.create(ctx, metadataConfig, tableBasePath, FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue)
-
- // TODO encoding should be done internally w/in HoodieBackedTableMetadata
- val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
-
- val recordsRDD: RDD[HoodieRecord[HoodieMetadataPayload]] =
- HoodieJavaRDD.getJavaRDD(
- metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)
- )
-
- val catalystRowsRDD: RDD[InternalRow] = recordsRDD.mapPartitions { it =>
- val metadataRecordSchema = new Parser().parse(metadataRecordSchemaString)
- val converter = AvroConversionUtils.createAvroToInternalRowConverter(metadataRecordSchema, metadataRecordStructType)
-
- it.map { record =>
- // schema and props are ignored for generating metadata record from the payload
- // instead, the underlying file system, or bloom filter, or columns stats metadata (part of payload) are directly used
- toScalaOption(record.getData.getInsertValue(null, null))
- .flatMap(avroRecord => converter(avroRecord.asInstanceOf[GenericRecord]))
- .orNull
- }
- }
+ val requiredIndexColumns =
+ targetColumnStatsIndexColumns.map(colName =>
+ col(s"${HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS}.${colName}"))
- HoodieUnsafeRDDUtils.createDataFrame(spark, catalystRowsRDD, metadataRecordStructType)
- }
- metadataTableDF
+ colStatsDF.where(col(HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS).isNotNull)
+ .select(requiredIndexColumns: _*)
}
}
object ColumnStatsIndexSupport {
- private val metadataRecordSchemaString: String = HoodieMetadataRecord.SCHEMA$.toString
- private val metadataRecordStructType: StructType = AvroConversionUtils.convertAvroSchemaToStructType(HoodieMetadataRecord.SCHEMA$)
+ private val columnStatsIndexProjectionSizeInMemoryThreshold = 100000
Review Comment:
could you explain a little why set `columnStatsIndexProjectionSizeInMemoryThreshold` to 100,000?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1190990714
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d3a78d17d2c9b747df8673042f08c218c6d0b331 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113)
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1154599306
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929357347
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +199,184 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
Review Comment:
Yes. At first, I thought the in-memory implementation is independent of Spark, but it is leveraging Spark's in-memory processing. I think the mode naming is still fine, just that `in-memory` mode can only be applied to Spark now.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1190885221
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
* d3a78d17d2c9b747df8673042f08c218c6d0b331 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] YannByron commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
YannByron commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928084578
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,18 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK = "spark";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
Review Comment:
maybe `.index.column.stats.processing.mode` is ok.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] YannByron commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
YannByron commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928087048
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java:
##########
@@ -170,7 +170,8 @@ Map<Pair<String, String>, HoodieMetadataColumnStats> getColumnStats(final List<P
* @return {@link HoodieData} of {@link HoodieRecord}s with records matching the passed in key prefixes.
*/
HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(List<String> keyPrefixes,
- String partitionName);
+ String partitionName,
+ boolean shouldLoadInMemory);
Review Comment:
can we declare a member variable named `shouldLoadColumnStatsInMemory` in `FileSystemBackedTableMetadata`, so that we don't need to change the definition of `getRecordsByKeyPrefixes`.
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java:
##########
@@ -170,7 +170,8 @@ Map<Pair<String, String>, HoodieMetadataColumnStats> getColumnStats(final List<P
* @return {@link HoodieData} of {@link HoodieRecord}s with records matching the passed in key prefixes.
*/
HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(List<String> keyPrefixes,
- String partitionName);
+ String partitionName,
+ boolean shouldLoadInMemory);
Review Comment:
can we declare a member variable named `shouldLoadColumnStatsInMemory` only in `FileSystemBackedTableMetadata`, so that we don't need to change the definition of `getRecordsByKeyPrefixes`.
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929282372
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,26 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE = "engine";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
+ .noDefaultValue()
+ .withValidValues(COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY, COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE)
+ .sinceVersion("0.12.0")
+ .withDocumentation("By default Column Stats Index is automatically determining whether it should be read and processed either"
+ + "'in-memory' (w/in executing process) or using Spark (on a cluster), based on some factors like the size of the Index "
+ + "and how many columns are read. This config allows to override this behavior.");
+
+ public static final ConfigProperty<Integer> COLUMN_STATS_INDEX_IN_MEMORY_PROJECTION_THRESHOLD = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.inMemory.projection.threshold")
+ .defaultValue(100000)
Review Comment:
Yes, reading Column Stats lesser than 100k rows is faster in-memory, than using Spark (AFAIR it was roughly about 1s vs 2s)
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +199,184 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
Review Comment:
All of the impl is Spark-specifc. To extend to other changes we will need to abstract quite a bit of functionality which we currently don't have any infra for (like `Expression`s, etc)
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +199,184 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
Review Comment:
Correct
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928153944
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
+ val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePathV2.toString)
// Read Metadata Table's Column Stats Index into Spark's [[DataFrame]]
- spark.read.format("org.apache.hudi")
+ val colStatsDF = spark.read.format("org.apache.hudi")
.options(metadataConfig.getProps.asScala)
.load(s"$metadataTablePath/${MetadataPartitionType.COLUMN_STATS.getPartitionPath}")
- }
-
- private def readColumnStatsIndexForColumnsInternal(spark: SparkSession, targetColumns: Seq[String], metadataConfig: HoodieMetadataConfig, tableBasePath: String) = {
- val ctx = new HoodieSparkEngineContext(new JavaSparkContext(spark.sparkContext))
- // Read Metadata Table's Column Stats Index into Spark's [[DataFrame]] by
- // - Fetching the records from CSI by key-prefixes (encoded column names)
- // - Deserializing fetched records into [[InternalRow]]s
- // - Composing [[DataFrame]]
- val metadataTableDF = {
- val metadataTable = HoodieTableMetadata.create(ctx, metadataConfig, tableBasePath, FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue)
-
- // TODO encoding should be done internally w/in HoodieBackedTableMetadata
- val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
-
- val recordsRDD: RDD[HoodieRecord[HoodieMetadataPayload]] =
- HoodieJavaRDD.getJavaRDD(
- metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)
- )
-
- val catalystRowsRDD: RDD[InternalRow] = recordsRDD.mapPartitions { it =>
- val metadataRecordSchema = new Parser().parse(metadataRecordSchemaString)
- val converter = AvroConversionUtils.createAvroToInternalRowConverter(metadataRecordSchema, metadataRecordStructType)
-
- it.map { record =>
- // schema and props are ignored for generating metadata record from the payload
- // instead, the underlying file system, or bloom filter, or columns stats metadata (part of payload) are directly used
- toScalaOption(record.getData.getInsertValue(null, null))
- .flatMap(avroRecord => converter(avroRecord.asInstanceOf[GenericRecord]))
- .orNull
- }
- }
+ val requiredIndexColumns =
+ targetColumnStatsIndexColumns.map(colName =>
+ col(s"${HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS}.${colName}"))
- HoodieUnsafeRDDUtils.createDataFrame(spark, catalystRowsRDD, metadataRecordStructType)
- }
- metadataTableDF
+ colStatsDF.where(col(HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS).isNotNull)
+ .select(requiredIndexColumns: _*)
}
}
object ColumnStatsIndexSupport {
- private val metadataRecordSchemaString: String = HoodieMetadataRecord.SCHEMA$.toString
- private val metadataRecordStructType: StructType = AvroConversionUtils.convertAvroSchemaToStructType(HoodieMetadataRecord.SCHEMA$)
+ private val columnStatsIndexProjectionSizeInMemoryThreshold = 100000
Review Comment:
Just an empirical estimate
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadata.java:
##########
@@ -170,7 +170,8 @@ Map<Pair<String, String>, HoodieMetadataColumnStats> getColumnStats(final List<P
* @return {@link HoodieData} of {@link HoodieRecord}s with records matching the passed in key prefixes.
*/
HoodieData<HoodieRecord<HoodieMetadataPayload>> getRecordsByKeyPrefixes(List<String> keyPrefixes,
- String partitionName);
+ String partitionName,
+ boolean shouldLoadInMemory);
Review Comment:
Good call. I think that would definitely make sense if we in consider in the future to make all methods be configurable to either load in-memory or using Spark. Right now only this method is configurable and it would be very misleading if we elevate it to be instance level config
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
Review Comment:
I don't think we currently have a need 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1194364604
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
}, {
"hash" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a7ba7493c4f35b74082d8d3bf66246feaf9a91a6 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248)
* b1e993b6f13c44a532a21470d0519329e2cd42be UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929358584
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,26 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE = "engine";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
+ .noDefaultValue()
+ .withValidValues(COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY, COLUMN_STATS_INDEX_PROCESSING_MODE_ENGINE)
+ .sinceVersion("0.12.0")
+ .withDocumentation("By default Column Stats Index is automatically determining whether it should be read and processed either"
+ + "'in-memory' (w/in executing process) or using Spark (on a cluster), based on some factors like the size of the Index "
+ + "and how many columns are read. This config allows to override this behavior.");
+
+ public static final ConfigProperty<Integer> COLUMN_STATS_INDEX_IN_MEMORY_PROJECTION_THRESHOLD = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.inMemory.projection.threshold")
+ .defaultValue(100000)
Review Comment:
How fast is it for in-memory?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1194527052
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
}, {
"hash" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10321",
"triggerID" : "b1e993b6f13c44a532a21470d0519329e2cd42be",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b1e993b6f13c44a532a21470d0519329e2cd42be Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10321)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] YannByron commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
YannByron commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928090091
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
Review Comment:
should this function need to support `in-memory` mode?
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193081655
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 909fb5e89b8e9e2e84a6f4892025ddb3715497f5 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226)
* a7ba7493c4f35b74082d8d3bf66246feaf9a91a6 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r928153607
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieMetadataConfig.java:
##########
@@ -187,6 +187,18 @@ public final class HoodieMetadataConfig extends HoodieConfig {
.sinceVersion("0.11.0")
.withDocumentation("Comma-separated list of columns for which column stats index will be built. If not set, all columns will be indexed");
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_IN_MEMORY = "in-memory";
+ public static final String COLUMN_STATS_INDEX_PROCESSING_MODE_SPARK = "spark";
+
+ public static final ConfigProperty<String> COLUMN_STATS_INDEX_PROCESSING_MODE_OVERRIDE = ConfigProperty
+ .key(METADATA_PREFIX + ".index.column.stats.processing.mode.override")
Review Comment:
Processing mode is not meant to be configured by default: this config is specifically to override this behavior
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929037411
##########
hudi-common/src/main/java/org/apache/hudi/BaseHoodieTableFileIndex.java:
##########
@@ -166,6 +167,11 @@ public Map<String, List<FileSlice>> listFileSlices() {
.collect(Collectors.toMap(e -> e.getKey().path, Map.Entry::getValue));
}
+ public int getFileSlicesCount() {
+ return cachedAllInputFileSlices.values().stream()
+ .reduce(0, (count, fileSlices) -> count + fileSlices.size(), Integer::sum);
Review Comment:
SG. Will take up as a follow-up
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929096225
##########
hudi-common/src/main/java/org/apache/hudi/BaseHoodieTableFileIndex.java:
##########
@@ -166,6 +167,11 @@ public Map<String, List<FileSlice>> listFileSlices() {
.collect(Collectors.toMap(e -> e.getKey().path, Map.Entry::getValue));
}
+ public int getFileSlicesCount() {
+ return cachedAllInputFileSlices.values().stream()
+ .reduce(0, (count, fileSlices) -> count + fileSlices.size(), Integer::sum);
Review Comment:
Addressed
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193101465
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* a7ba7493c4f35b74082d8d3bf66246feaf9a91a6 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1154601446
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1190922482
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272)
* d3a78d17d2c9b747df8673042f08c218c6d0b331 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113)
* c9d246f0080064b4d2022a64df7e3d2cb13718d2 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193082230
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
}, {
"hash" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248",
"triggerID" : "a7ba7493c4f35b74082d8d3bf66246feaf9a91a6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 909fb5e89b8e9e2e84a6f4892025ddb3715497f5 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10226)
* a7ba7493c4f35b74082d8d3bf66246feaf9a91a6 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10248)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1193012776
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=9272",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
}, {
"hash" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10113",
"triggerID" : "d3a78d17d2c9b747df8673042f08c218c6d0b331",
"triggerType" : "PUSH"
}, {
"hash" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10115",
"triggerID" : "c9d246f0080064b4d2022a64df7e3d2cb13718d2",
"triggerType" : "PUSH"
}, {
"hash" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164",
"triggerID" : "0a15f230f412c1670dbf32465e99fe47af77c066",
"triggerType" : "PUSH"
}, {
"hash" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "909fb5e89b8e9e2e84a6f4892025ddb3715497f5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0a15f230f412c1670dbf32465e99fe47af77c066 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=10164)
* 909fb5e89b8e9e2e84a6f4892025ddb3715497f5 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929075635
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/ColumnStatsIndexSupport.scala:
##########
@@ -112,154 +195,186 @@ trait ColumnStatsIndexSupport extends SparkAdapterSupport {
* column references from the filtering expressions, and only transpose records corresponding to the
* columns referenced in those
*
- * @param spark Spark session ref
- * @param colStatsDF [[DataFrame]] bearing raw Column Stats Index table
+ * @param colStatsRecords [[HoodieData[HoodieMetadataColumnStats]]] bearing raw Column Stats Index records
* @param queryColumns target columns to be included into the final table
- * @param tableSchema schema of the source data table
* @return reshaped table according to the format outlined above
*/
- def transposeColumnStatsIndex(spark: SparkSession, colStatsDF: DataFrame, queryColumns: Seq[String], tableSchema: StructType): DataFrame = {
- val colStatsSchema = colStatsDF.schema
- val colStatsSchemaOrdinalsMap = colStatsSchema.fields.zipWithIndex.map({
- case (field, ordinal) => (field.name, ordinal)
- }).toMap
-
+ private def transpose(colStatsRecords: HoodieData[HoodieMetadataColumnStats], queryColumns: Seq[String]): (HoodieData[Row], StructType) = {
val tableSchemaFieldMap = tableSchema.fields.map(f => (f.name, f)).toMap
-
- val colNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_COLUMN_NAME)
- val minValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MIN_VALUE)
- val maxValueOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_MAX_VALUE)
- val fileNameOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_FILE_NAME)
- val nullCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_NULL_COUNT)
- val valueCountOrdinal = colStatsSchemaOrdinalsMap(HoodieMetadataPayload.COLUMN_STATS_FIELD_VALUE_COUNT)
-
- // NOTE: We have to collect list of indexed columns to make sure we properly align the rows
- // w/in the transposed dataset: since some files might not have all of the columns indexed
- // either due to the Column Stats Index config changes, schema evolution, etc, we have
- // to make sure that all of the rows w/in transposed data-frame are properly padded (with null
- // values) for such file-column combinations
- val indexedColumns: Seq[String] = colStatsDF.rdd.map(row => row.getString(colNameOrdinal)).distinct().collect()
-
// NOTE: We're sorting the columns to make sure final index schema matches layout
// of the transposed table
- val sortedTargetColumns = TreeSet(queryColumns.intersect(indexedColumns): _*)
-
- val transposedRDD = colStatsDF.rdd
- .filter(row => sortedTargetColumns.contains(row.getString(colNameOrdinal)))
- .map { row =>
- if (row.isNullAt(minValueOrdinal) && row.isNullAt(maxValueOrdinal)) {
+ val sortedTargetColumnsSet = TreeSet(queryColumns:_*)
+ val sortedTargetColumns = sortedTargetColumnsSet.toSeq
+
+ // NOTE: This is a trick to avoid pulling all of [[ColumnStatsIndexSupport]] object into the lambdas'
+ // closures below
+ val indexedColumns = this.indexedColumns
+
+ // Here we perform complex transformation which requires us to modify the layout of the rows
+ // of the dataset, and therefore we rely on low-level RDD API to avoid incurring encoding/decoding
+ // penalty of the [[Dataset]], since it's required to adhere to its schema at all times, while
+ // RDDs are not;
+ val transposedRows: HoodieData[Row] = colStatsRecords
+ // NOTE: Explicit conversion is required for Scala 2.11
+ .filter(JFunction.toJavaSerializableFunction(r => sortedTargetColumnsSet.contains(r.getColumnName)))
+ .mapToPair(JFunction.toJavaSerializablePairFunction(r => {
+ if (r.getMinValue == null && r.getMaxValue == null) {
// Corresponding row could be null in either of the 2 cases
// - Column contains only null values (in that case both min/max have to be nulls)
// - This is a stubbed Column Stats record (used as a tombstone)
- row
+ collection.Pair.of(r.getFileName, r)
} else {
- val minValueStruct = row.getAs[Row](minValueOrdinal)
- val maxValueStruct = row.getAs[Row](maxValueOrdinal)
+ val minValueWrapper = r.getMinValue
+ val maxValueWrapper = r.getMaxValue
- checkState(minValueStruct != null && maxValueStruct != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
+ checkState(minValueWrapper != null && maxValueWrapper != null, "Invalid Column Stats record: either both min/max have to be null, or both have to be non-null")
- val colName = row.getString(colNameOrdinal)
+ val colName = r.getColumnName
val colType = tableSchemaFieldMap(colName).dataType
- val (minValue, _) = tryUnpackNonNullVal(minValueStruct)
- val (maxValue, _) = tryUnpackNonNullVal(maxValueStruct)
- val rowValsSeq = row.toSeq.toArray
+ val minValue = deserialize(tryUnpackValueWrapper(minValueWrapper), colType)
+ val maxValue = deserialize(tryUnpackValueWrapper(maxValueWrapper), colType)
+
// Update min-/max-value structs w/ unwrapped values in-place
- rowValsSeq(minValueOrdinal) = deserialize(minValue, colType)
- rowValsSeq(maxValueOrdinal) = deserialize(maxValue, colType)
+ r.setMinValue(minValue)
+ r.setMaxValue(maxValue)
- Row(rowValsSeq: _*)
+ collection.Pair.of(r.getFileName, r)
}
- }
- .groupBy(r => r.getString(fileNameOrdinal))
- .foldByKey(Seq[Row]()) {
- case (_, columnRowsSeq) =>
- // Rows seq is always non-empty (otherwise it won't be grouped into)
- val fileName = columnRowsSeq.head.get(fileNameOrdinal)
- val valueCount = columnRowsSeq.head.get(valueCountOrdinal)
-
- // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
- // to align existing column-stats for individual file with the list of expected ones for the
- // whole transposed projection (a superset of all files)
- val columnRowsMap = columnRowsSeq.map(row => (row.getString(colNameOrdinal), row)).toMap
- val alignedColumnRowsSeq = sortedTargetColumns.toSeq.map(columnRowsMap.get)
-
- val coalescedRowValuesSeq =
- alignedColumnRowsSeq.foldLeft(Seq[Any](fileName, valueCount)) {
- case (acc, opt) =>
- opt match {
- case Some(columnStatsRow) =>
- acc ++ Seq(minValueOrdinal, maxValueOrdinal, nullCountOrdinal).map(ord => columnStatsRow.get(ord))
- case None =>
- // NOTE: Since we're assuming missing column to essentially contain exclusively
- // null values, we set null-count to be equal to value-count (this behavior is
- // consistent with reading non-existent columns from Parquet)
- acc ++ Seq(null, null, valueCount)
- }
- }
-
- Seq(Row(coalescedRowValuesSeq:_*))
- }
- .values
- .flatMap(it => it)
+ }))
+ .groupByKey()
+ .map(JFunction.toJavaSerializableFunction(p => {
+ val columnRecordsSeq: Seq[HoodieMetadataColumnStats] = p.getValue.asScala.toSeq
+ val fileName: String = p.getKey
+ val valueCount: Long = columnRecordsSeq.head.getValueCount
+
+ // To properly align individual rows (corresponding to a file) w/in the transposed projection, we need
+ // to align existing column-stats for individual file with the list of expected ones for the
+ // whole transposed projection (a superset of all files)
+ val columnRecordsMap = columnRecordsSeq.map(r => (r.getColumnName, r)).toMap
+ val alignedColStatRecordsSeq = sortedTargetColumns.map(columnRecordsMap.get)
+
+ val coalescedRowValuesSeq =
+ alignedColStatRecordsSeq.foldLeft(ListBuffer[Any](fileName, valueCount)) {
+ case (acc, opt) =>
+ opt match {
+ case Some(colStatRecord) =>
+ acc ++= Seq(colStatRecord.getMinValue, colStatRecord.getMaxValue, colStatRecord.getNullCount)
+ case None =>
+ // NOTE: This could occur in either of the following cases:
+ // 1. Column is not indexed in Column Stats Index: in this case we won't be returning
+ // any statistics for such column (ie all stats will be null)
+ // 2. Particular file does not have this particular column (which is indexed by Column Stats Index):
+ // in this case we're assuming missing column to essentially contain exclusively
+ // null values, we set min/max values as null and null-count to be equal to value-count (this
+ // behavior is consistent with reading non-existent columns from Parquet)
+ //
+ // This is a way to determine current column's index without explicit iteration (we're adding 3 stats / column)
+ val idx = acc.length / 3
+ val colName = sortedTargetColumns(idx)
+ val indexed = indexedColumns.contains(colName)
+
+ val nullCount = if (indexed) valueCount else null
+
+ acc ++= Seq(null, null, nullCount)
+ }
+ }
+
+ Row(coalescedRowValuesSeq:_*)
+ }))
// NOTE: It's crucial to maintain appropriate ordering of the columns
// matching table layout: hence, we cherry-pick individual columns
// instead of simply filtering in the ones we're interested in the schema
- val indexSchema = composeIndexSchema(sortedTargetColumns.toSeq, tableSchema)
+ val indexSchema = composeIndexSchema(sortedTargetColumns, tableSchema)
+ (transposedRows, indexSchema)
+ }
- spark.createDataFrame(transposedRDD, indexSchema)
+ private def loadColumnStatsIndexForColumnsInternal(targetColumns: Seq[String], shouldReadInMemory: Boolean): DataFrame = {
+ val colStatsDF = {
+ val colStatsRecords: HoodieData[HoodieMetadataColumnStats] = loadColumnStatsIndexRecords(targetColumns, shouldReadInMemory)
+ // NOTE: Explicit conversion is required for Scala 2.11
+ val catalystRows: HoodieData[InternalRow] = colStatsRecords.mapPartitions(JFunction.toJavaSerializableFunction(it => {
+ val converter = AvroConversionUtils.createAvroToInternalRowConverter(HoodieMetadataColumnStats.SCHEMA$, columnStatsRecordStructType)
+ it.asScala.map(r => converter(r).orNull).asJava
+ }), false)
+
+ if (shouldReadInMemory) {
+ // NOTE: This will instantiate a [[Dataset]] backed by [[LocalRelation]] holding all of the rows
+ // of the transposed table in memory, facilitating execution of the subsequently chained operations
+ // on it locally (on the driver; all such operations are actually going to be performed by Spark's
+ // Optimizer)
+ createDataFrameFromInternalRows(spark, catalystRows.collectAsList().asScala, columnStatsRecordStructType)
+ } else {
+ createDataFrameFromRDD(spark, HoodieJavaRDD.getJavaRDD(catalystRows), columnStatsRecordStructType)
+ }
+ }
+
+ colStatsDF.select(targetColumnStatsIndexColumns.map(col): _*)
+ }
+
+ private def loadColumnStatsIndexRecords(targetColumns: Seq[String], shouldReadInMemory: Boolean): HoodieData[HoodieMetadataColumnStats] = {
+ // Read Metadata Table's Column Stats Index records into [[HoodieData]] container by
+ // - Fetching the records from CSI by key-prefixes (encoded column names)
+ // - Extracting [[HoodieMetadataColumnStats]] records
+ // - Filtering out nulls
+ checkState(targetColumns.nonEmpty)
+
+ // TODO encoding should be done internally w/in HoodieBackedTableMetadata
+ val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
+
+ val metadataRecords: HoodieData[HoodieRecord[HoodieMetadataPayload]] =
+ metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS, shouldReadInMemory)
+
+ val columnStatsRecords: HoodieData[HoodieMetadataColumnStats] =
+ // NOTE: Explicit conversion is required for Scala 2.11
+ metadataRecords.map(JFunction.toJavaSerializableFunction(record => {
+ toScalaOption(record.getData.getInsertValue(null, null))
+ .map(metadataRecord => metadataRecord.asInstanceOf[HoodieMetadataRecord].getColumnStatsMetadata)
+ .orNull
+ }))
+ .filter(JFunction.toJavaSerializableFunction(columnStatsRecord => columnStatsRecord != null))
+
+ columnStatsRecords
}
- private def readFullColumnStatsIndexInternal(spark: SparkSession, metadataConfig: HoodieMetadataConfig, tableBasePath: String): DataFrame = {
- val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(tableBasePath)
+ private def loadFullColumnStatsIndexInternal(): DataFrame = {
+ val metadataTablePath = HoodieTableMetadata.getMetadataTableBasePath(metaClient.getBasePathV2.toString)
// Read Metadata Table's Column Stats Index into Spark's [[DataFrame]]
- spark.read.format("org.apache.hudi")
+ val colStatsDF = spark.read.format("org.apache.hudi")
.options(metadataConfig.getProps.asScala)
.load(s"$metadataTablePath/${MetadataPartitionType.COLUMN_STATS.getPartitionPath}")
- }
-
- private def readColumnStatsIndexForColumnsInternal(spark: SparkSession, targetColumns: Seq[String], metadataConfig: HoodieMetadataConfig, tableBasePath: String) = {
- val ctx = new HoodieSparkEngineContext(new JavaSparkContext(spark.sparkContext))
- // Read Metadata Table's Column Stats Index into Spark's [[DataFrame]] by
- // - Fetching the records from CSI by key-prefixes (encoded column names)
- // - Deserializing fetched records into [[InternalRow]]s
- // - Composing [[DataFrame]]
- val metadataTableDF = {
- val metadataTable = HoodieTableMetadata.create(ctx, metadataConfig, tableBasePath, FileSystemViewStorageConfig.SPILLABLE_DIR.defaultValue)
-
- // TODO encoding should be done internally w/in HoodieBackedTableMetadata
- val encodedTargetColumnNames = targetColumns.map(colName => new ColumnIndexID(colName).asBase64EncodedString())
-
- val recordsRDD: RDD[HoodieRecord[HoodieMetadataPayload]] =
- HoodieJavaRDD.getJavaRDD(
- metadataTable.getRecordsByKeyPrefixes(encodedTargetColumnNames.asJava, HoodieTableMetadataUtil.PARTITION_NAME_COLUMN_STATS)
- )
-
- val catalystRowsRDD: RDD[InternalRow] = recordsRDD.mapPartitions { it =>
- val metadataRecordSchema = new Parser().parse(metadataRecordSchemaString)
- val converter = AvroConversionUtils.createAvroToInternalRowConverter(metadataRecordSchema, metadataRecordStructType)
-
- it.map { record =>
- // schema and props are ignored for generating metadata record from the payload
- // instead, the underlying file system, or bloom filter, or columns stats metadata (part of payload) are directly used
- toScalaOption(record.getData.getInsertValue(null, null))
- .flatMap(avroRecord => converter(avroRecord.asInstanceOf[GenericRecord]))
- .orNull
- }
- }
+ val requiredIndexColumns =
+ targetColumnStatsIndexColumns.map(colName =>
+ col(s"${HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS}.${colName}"))
- HoodieUnsafeRDDUtils.createDataFrame(spark, catalystRowsRDD, metadataRecordStructType)
- }
- metadataTableDF
+ colStatsDF.where(col(HoodieMetadataPayload.SCHEMA_FIELD_ID_COLUMN_STATS).isNotNull)
+ .select(requiredIndexColumns: _*)
}
}
object ColumnStatsIndexSupport {
- private val metadataRecordSchemaString: String = HoodieMetadataRecord.SCHEMA$.toString
- private val metadataRecordStructType: StructType = AvroConversionUtils.convertAvroSchemaToStructType(HoodieMetadataRecord.SCHEMA$)
+ private val columnStatsIndexProjectionSizeInMemoryThreshold = 100000
Review Comment:
I've tried to specifically avoid exposing such config, but now am thinking that we probably should allow configuring 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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua commented on a diff in pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua commented on code in PR #5746:
URL: https://github.com/apache/hudi/pull/5746#discussion_r929361712
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCatalystUtils.scala:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hudi
+
+import org.apache.hudi.common.data.HoodieData
+import org.apache.spark.sql.Dataset
+import org.apache.spark.storage.StorageLevel
+import org.apache.spark.storage.StorageLevel._
+
+object HoodieCatalystUtils extends SparkAdapterSupport {
+
+ /**
+ * Executes provided function while keeping provided [[Dataset]] instance persisted for the
+ * duration of the execution
+ *
+ * @param df target [[Dataset]] to be persisted
+ * @param level desired [[StorageLevel]] of the persistence
+ * @param f target function to be executed while [[Dataset]] is kept persisted
+ * @tparam T return value of the target function
+ * @return execution outcome of the [[f]] function
+ */
+ def withPersistedDataset[T](df: Dataset[_], level: StorageLevel = MEMORY_AND_DISK)(f: => T): T = {
+ df.persist(level)
+ try {
+ f
+ } finally {
+ df.unpersist()
+ }
Review Comment:
So this is moved from `HoodieDatasetUtils`
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] yihua merged pull request #5746: [HUDI-4250][HUDI-4202] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
yihua merged PR #5746:
URL: https://github.com/apache/hudi/pull/5746
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] alexeykudinkin commented on pull request #5746: [WIP] Optimize performance of Column Stats filtering seq
Posted by GitBox <gi...@apache.org>.
alexeykudinkin commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1149246115
Build had succeeded:
https://dev.azure.com/apache-hudi-ci-org/apache-hudi-ci/_build/results?buildId=9137&view=results
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #5746: [HUDI-4250][HUDI-4202][Stacked on 5855] Optimize performance of Column Stats Index reading in Data Skipping
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #5746:
URL: https://github.com/apache/hudi/pull/5746#issuecomment-1154597411
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5a2d1f4b1469297a51ae0b07a5982383658beaf9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5a2d1f4b1469297a51ae0b07a5982383658beaf9 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
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: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org