You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "jonvex (via GitHub)" <gi...@apache.org> on 2024/06/07 02:33:48 UTC
[PR] [HUDI-7840] Add position merging to fg reader [hudi]
jonvex opened a new pull request, #11413:
URL: https://github.com/apache/hudi/pull/11413
### Change Logs
Position merging was removed in [HUDI-7567](https://github.com/apache/hudi/pull/10957)
but will now be added back with additional testing and proper use of filters. As well as positional merge for bootstrap
### Impact
positional merge feature added back to fg reader
### Risk level (write none, low medium or high below)
medium
wrote tests
### Documentation Update
N/A
### Contributor's checklist
- [ ] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [ ] Change Logs and Impact were stated clearly
- [ ] Adequate tests were added if applicable
- [ ] CI passed
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631737729
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestHoodiePositionBasedFileGroupRecordBuffer.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer;
+import org.apache.hudi.common.table.read.HoodiePositionBasedSchemaHandler;
+import org.apache.hudi.common.table.read.TestHoodieFileGroupReaderOnSpark;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.SchemaTestUtil;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.ExternalSpillableMap;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieValidationException;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.engine.HoodieReaderContext.INTERNAL_META_RECORD_KEY;
+import static org.apache.hudi.common.model.WriteOperationType.INSERT;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.createMetaClient;
+import static org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestHoodiePositionBasedFileGroupRecordBuffer extends TestHoodieFileGroupReaderOnSpark {
+ private final HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(0xDEEF);
+ private HoodieTableMetaClient metaClient;
+ private Schema avroSchema;
+ private HoodiePositionBasedFileGroupRecordBuffer<InternalRow> buffer;
+ private String partitionPath;
+
+ public void prepareBuffer(boolean useCustomMerger) throws Exception {
+ Map<String, String> writeConfigs = new HashMap<>();
+ writeConfigs.put(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet");
+ writeConfigs.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+ writeConfigs.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partition_path");
+ writeConfigs.put("hoodie.datasource.write.precombine.field", "timestamp");
+ writeConfigs.put("hoodie.payload.ordering.field", "timestamp");
+ writeConfigs.put(HoodieTableConfig.HOODIE_TABLE_NAME_KEY, "hoodie_test");
+ writeConfigs.put("hoodie.insert.shuffle.parallelism", "4");
+ writeConfigs.put("hoodie.upsert.shuffle.parallelism", "4");
+ writeConfigs.put("hoodie.bulkinsert.shuffle.parallelism", "2");
+ writeConfigs.put("hoodie.delete.shuffle.parallelism", "1");
+ writeConfigs.put("hoodie.merge.small.file.group.candidates.limit", "0");
+ writeConfigs.put("hoodie.compact.inline", "false");
+ writeConfigs.put(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), "true");
+ commitToTable(recordsToStrings(dataGen.generateInserts("001", 100)), INSERT.value(), writeConfigs);
+
+ String[] partitionPaths = dataGen.getPartitionPaths();
+ String[] partitionValues = new String[1];
+ partitionPath = partitionPaths[0];
+ partitionValues[0] = partitionPath;
+
+ metaClient = createMetaClient(getStorageConf(), getBasePath());
+ avroSchema = new TableSchemaResolver(metaClient).getTableAvroSchema();
+ Option<String[]> partitionFields = metaClient.getTableConfig().getPartitionFields();
+ Option<String> partitionNameOpt = StringUtils.isNullOrEmpty(partitionPaths[0])
+ ? Option.empty() : Option.of(partitionPaths[0]);
+
+ HoodieReaderContext ctx = getHoodieReaderContext(getBasePath(), avroSchema, getStorageConf());
+ ctx.setTablePath(metaClient.getBasePathV2().toString());
+ ctx.setLatestCommitTime(metaClient.createNewInstantTime());
+ ctx.setShouldMergeUseRecordPosition(true);
+ ctx.setHasBootstrapBaseFile(false);
+ ctx.setHasLogFiles(true);
+ ctx.setNeedsBootstrapMerge(false);
Review Comment:
Yes covered in TestFiltersInFileGroupReader and also some existing bootstrap tests should cover 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155650961
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
}, {
"hash" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24280",
"triggerID" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"triggerType" : "PUSH"
}, {
"hash" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 1ce1d753818efb6be00c20fb5a8dd141c7c47f00 Azure: [CANCELED](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24280)
* 4d4c0fdc03b72cfb7ad86172a75fcca439e42682 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2153783515
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631704549
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
+ //we need to iterate row by row when merging
+ new CachingIterator[InternalRow] {
+ val combinedRow = new JoinedRow()
+
+ //position column will always be at the end of the row
+ private def getPos(row: InternalRow): Long = {
+ row.getLong(row.numFields-1)
+ }
+
+ private def getNextSkeleton: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ private def getNextData: (InternalRow, Long) = {
+ val nextDataRow = dataFileIterator.next().asInstanceOf[InternalRow]
+ (nextDataRow, getPos(nextDataRow))
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+
+ override protected def doHasNext(): Boolean = {
+ if (!dataFileIterator.hasNext || !skeletonFileIterator.hasNext) {
+ false
+ } else {
+ var nextSkeleton = getNextSkeleton
+ var nextData = getNextData
+ while (nextSkeleton._2 != nextData._2) {
+ if (nextSkeleton._2 > nextData._2) {
+ if (!dataFileIterator.hasNext) {
+ return false
+ } else {
+ nextData = getNextData
+ }
} else {
- vecs(i) = d.column(i - s.numCols())
+ if (!skeletonFileIterator.hasNext) {
+ return false
+ } else {
+ nextSkeleton = getNextSkeleton
+ }
}
}
- assert(s.numRows() == d.numRows())
- sparkAdapter.makeColumnarBatch(vecs, s.numRows())
- case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ nextRecord = combinedRow(skeletonProjection.apply(nextSkeleton._1), dataProjection.apply(nextData._1))
+ true
+ }
}
}
+ } else {
+ new ClosableIterator[Any] {
+ val combinedRow = new JoinedRow()
- override def close(): Unit = {
- skeletonFileIterator.close()
- dataFileIterator.close()
- }
- }.asInstanceOf[ClosableIterator[InternalRow]]
+ override def hasNext: Boolean = {
+ //If the iterators are out of sync it is probably due to filter pushdown
+ checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
+ "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
+ dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ }
+
+ override def next(): Any = {
+ (skeletonFileIterator.next(), dataFileIterator.next()) match {
+ case (s: ColumnarBatch, d: ColumnarBatch) =>
+ //This will not be used until [HUDI-7693] is implemented
+ val numCols = s.numCols() + d.numCols()
+ val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
+ for (i <- 0 until numCols) {
+ if (i < s.numCols()) {
+ vecs(i) = s.column(i)
+ } else {
+ vecs(i) = d.column(i - s.numCols())
+ }
+ }
+ assert(s.numRows() == d.numRows())
+ sparkAdapter.makeColumnarBatch(vecs, s.numRows())
+ case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ }
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+ }.asInstanceOf[ClosableIterator[InternalRow]]
+ }
}
}
+
+object SparkFileFormatInternalRowReaderContext {
+ // From "namedExpressions.scala": Used to construct to record position field metadata.
+ private val FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY = "__file_source_generated_metadata_col"
+ private val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ private val METADATA_COL_ATTR_KEY = "__metadata_col"
+
+ def getAppliedRequiredSchema(requiredSchema: StructType, shouldAddRecordPosition: Boolean): StructType = {
+ if (shouldAddRecordPosition) {
+ val metadata = new MetadataBuilder()
+ .putString(METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .putString(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .build()
+ val rowIndexField = StructField(ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = false, metadata)
+ StructType(requiredSchema.fields.filterNot(isIndexTempColumn) :+ rowIndexField)
+ } else {
+ requiredSchema
+ }
+ }
+
+ /**
+ * Only valid if there is support for record positions and no log files
Review Comment:
`there is support for record positions` -> does this refer to reading row index in the Spark parquet reader?
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -56,16 +60,27 @@ import scala.collection.mutable
*/
class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetReader,
recordKeyColumn: String,
- filters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
+ filters: Seq[Filter],
+ requiredFilters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
Review Comment:
nit: add docs on the parameter `requiredFilters`
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -56,16 +60,27 @@ import scala.collection.mutable
*/
class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetReader,
recordKeyColumn: String,
- filters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
+ filters: Seq[Filter],
+ requiredFilters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
lazy val sparkAdapter: SparkAdapter = SparkAdapterSupport.sparkAdapter
+ private lazy val bootstrapSafeFilters: Seq[Filter] = filters.filter(filterIsSafeForBootstrap) ++ requiredFilters
private val deserializerMap: mutable.Map[Schema, HoodieAvroDeserializer] = mutable.Map()
+ private lazy val allFilters = filters ++ requiredFilters
+
+ override def supportsPositionField: Boolean = {
Review Comment:
nit: `supportsPositionField` -> `supportsParquetRowIndex`
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -56,16 +60,27 @@ import scala.collection.mutable
*/
class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetReader,
recordKeyColumn: String,
- filters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
+ filters: Seq[Filter],
+ requiredFilters: Seq[Filter]) extends BaseSparkInternalRowReaderContext {
lazy val sparkAdapter: SparkAdapter = SparkAdapterSupport.sparkAdapter
+ private lazy val bootstrapSafeFilters: Seq[Filter] = filters.filter(filterIsSafeForBootstrap) ++ requiredFilters
private val deserializerMap: mutable.Map[Schema, HoodieAvroDeserializer] = mutable.Map()
+ private lazy val allFilters = filters ++ requiredFilters
+
+ override def supportsPositionField: Boolean = {
+ HoodieSparkUtils.gteqSpark3_5
+ }
override def getFileRecordIterator(filePath: StoragePath,
start: Long,
length: Long,
dataSchema: Schema,
requiredSchema: Schema,
storage: HoodieStorage): ClosableIterator[InternalRow] = {
+ val hasPositionField = AvroSchemaUtils.containsFieldInSchema(requiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME)
Review Comment:
```suggestion
val hasRowIndexField = AvroSchemaUtils.containsFieldInSchema(requiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME)
```
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/execution/datasources/parquet/TestSparkFileFormatInternalRowReaderContext.scala:
##########
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.execution.datasources.parquet
+
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext.filterIsSafeForBootstrap
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
+import org.apache.spark.sql.sources.{And, IsNotNull, Or}
+import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.junit.jupiter.api.Test
+
+class TestSparkFileFormatInternalRowReaderContext extends SparkClientFunctionalTestHarness {
Review Comment:
Note: `TestHoodieFileGroupReaderBasedParquetFileFormat` is removed in #10957 and renamed to this class in this PR.
--
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
Re: [PR] [HUDI-7840] Add position merging to the new file group reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155656514
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
}, {
"hash" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24280",
"triggerID" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"triggerType" : "PUSH"
}, {
"hash" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24281",
"triggerID" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"triggerType" : "PUSH"
}, {
"hash" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4d4c0fdc03b72cfb7ad86172a75fcca439e42682 Azure: [CANCELED](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24281)
* 6c15d7a0558284728296d74c9acbc6805230d9a2 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
Re: [PR] [HUDI-7840] Add position merging to the new file group reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #11413:
URL: https://github.com/apache/hudi/pull/11413
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155239340
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4e6335c7cfb18881776d572954558a41aa33b91d Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273)
<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
Re: [PR] [HUDI-7840] Add position merging to the new file group reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155691424
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
}, {
"hash" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24280",
"triggerID" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"triggerType" : "PUSH"
}, {
"hash" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24281",
"triggerID" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"triggerType" : "PUSH"
}, {
"hash" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24283",
"triggerID" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"triggerType" : "PUSH"
}, {
"hash" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24285",
"triggerID" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"triggerType" : "PUSH"
}, {
"hash" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24284",
"triggerID" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 0bf72cfded469e3cc1091827cbe4f2f3c16de830 Azure: [CANCELED](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24285) Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24284)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631711150
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
Review Comment:
We can still iterate through rows within the `ColumnarBatch` in the vectorized processing. We can leave that as a follow-up.
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java:
##########
@@ -122,6 +123,15 @@ public void setNeedsBootstrapMerge(boolean needsBootstrapMerge) {
this.needsBootstrapMerge = needsBootstrapMerge;
}
+ // Getter and Setter for useRecordPosition
+ public boolean getUseRecordPosition() {
+ return useRecordPosition;
+ }
Review Comment:
Rename the getter and setter to sth like `shouldMergeUseRecordPosition` and `setMergeUseRecordPosition` so it indicates this is used for controlling the merging behavior.
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
+ //we need to iterate row by row when merging
+ new CachingIterator[InternalRow] {
+ val combinedRow = new JoinedRow()
+
+ //position column will always be at the end of the row
+ private def getPos(row: InternalRow): Long = {
+ row.getLong(row.numFields-1)
+ }
+
+ private def getNextSkeleton: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ private def getNextData: (InternalRow, Long) = {
+ val nextDataRow = dataFileIterator.next().asInstanceOf[InternalRow]
+ (nextDataRow, getPos(nextDataRow))
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+
+ override protected def doHasNext(): Boolean = {
+ if (!dataFileIterator.hasNext || !skeletonFileIterator.hasNext) {
+ false
+ } else {
+ var nextSkeleton = getNextSkeleton
+ var nextData = getNextData
+ while (nextSkeleton._2 != nextData._2) {
+ if (nextSkeleton._2 > nextData._2) {
+ if (!dataFileIterator.hasNext) {
+ return false
+ } else {
+ nextData = getNextData
+ }
} else {
- vecs(i) = d.column(i - s.numCols())
+ if (!skeletonFileIterator.hasNext) {
+ return false
+ } else {
+ nextSkeleton = getNextSkeleton
+ }
}
}
- assert(s.numRows() == d.numRows())
- sparkAdapter.makeColumnarBatch(vecs, s.numRows())
- case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ nextRecord = combinedRow(skeletonProjection.apply(nextSkeleton._1), dataProjection.apply(nextData._1))
+ true
+ }
}
}
+ } else {
+ new ClosableIterator[Any] {
+ val combinedRow = new JoinedRow()
- override def close(): Unit = {
- skeletonFileIterator.close()
- dataFileIterator.close()
- }
- }.asInstanceOf[ClosableIterator[InternalRow]]
+ override def hasNext: Boolean = {
+ //If the iterators are out of sync it is probably due to filter pushdown
+ checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
+ "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
+ dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ }
+
+ override def next(): Any = {
+ (skeletonFileIterator.next(), dataFileIterator.next()) match {
Review Comment:
I assume this is the original logic of key-based merging which is unchanged.
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155337873
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4e6335c7cfb18881776d572954558a41aa33b91d Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273)
* d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631737122
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java:
##########
@@ -301,9 +311,19 @@ public final UnaryOperator<T> projectRecord(Schema from, Schema to) {
* @return the record position in the base file.
*/
public long extractRecordPosition(T record, Schema schema, String fieldName, long providedPositionIfNeeded) {
+ if (supportsParquetRowIndex()) {
+ Object position = getValue(record, schema, fieldName);
+ if (position != null) {
+ return (long) position;
+ }
Review Comment:
sure
--
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
Re: [PR] [HUDI-7840] Add position merging to the new file group reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155701487
Azure CI is green.
<img width="1379" alt="Screenshot 2024-06-07 at 16 46 25" src="https://github.com/apache/hudi/assets/2497195/2bf57a53-50c2-4881-b09e-b9d8025c058a">
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155328163
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4e6335c7cfb18881776d572954558a41aa33b91d Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273)
* d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631715014
##########
hudi-common/src/main/java/org/apache/hudi/common/table/read/HoodiePositionBasedSchemaHandler.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.common.table.read;
+
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.internal.schema.InternalSchema;
+
+import org.apache.avro.Schema;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.hudi.avro.AvroSchemaUtils.appendFieldsToSchemaDedupNested;
+
+/**
+ * This class is responsible for handling the schema for the file group reader that supports positional merge.
+ */
+public class HoodiePositionBasedSchemaHandler<T> extends HoodieFileGroupReaderSchemaHandler<T> {
+ public HoodiePositionBasedSchemaHandler(HoodieReaderContext<T> readerContext,
+ Schema dataSchema,
+ Schema requestedSchema,
+ Option<InternalSchema> internalSchemaOpt,
+ HoodieTableConfig hoodieTableConfig) {
+ super(readerContext, dataSchema, requestedSchema, internalSchemaOpt, hoodieTableConfig);
+
Review Comment:
nit: remove empty line.
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/execution/datasources/parquet/TestSparkFileFormatInternalRowReaderContext.scala:
##########
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.execution.datasources.parquet
+
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext.filterIsSafeForBootstrap
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
+import org.apache.spark.sql.sources.{And, IsNotNull, Or}
+import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.junit.jupiter.api.Test
+
+class TestSparkFileFormatInternalRowReaderContext extends SparkClientFunctionalTestHarness {
+
+ @Test
+ def testBootstrapFilters(): Unit = {
+ val recordKeyField = HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD.getFieldName
+ val commitTimeField = HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD.getFieldName
+
+ val recordKeyFilter = IsNotNull(recordKeyField)
+ assertTrue(filterIsSafeForBootstrap(recordKeyFilter))
+ val commitTimeFilter = IsNotNull(commitTimeField)
+ assertTrue(filterIsSafeForBootstrap(commitTimeFilter))
+
+ val dataFieldFilter = IsNotNull("someotherfield")
+ assertTrue(filterIsSafeForBootstrap(dataFieldFilter))
+
+ val legalComplexFilter = Or(recordKeyFilter, commitTimeFilter)
+ assertTrue(filterIsSafeForBootstrap(legalComplexFilter))
+
+ val illegalComplexFilter = Or(recordKeyFilter, dataFieldFilter)
+ assertFalse(filterIsSafeForBootstrap(illegalComplexFilter))
+
+ val illegalNestedFilter = And(legalComplexFilter, illegalComplexFilter)
+ assertFalse(filterIsSafeForBootstrap(illegalNestedFilter))
+
+ val legalNestedFilter = And(legalComplexFilter, recordKeyFilter)
+ assertTrue(filterIsSafeForBootstrap(legalNestedFilter))
+ }
+
+ @Test
+ def testGetAppliedRequiredSchema(): Unit = {
+ val fields = Array(
+ StructField("column_a", LongType, nullable = false),
+ StructField("column_b", StringType, nullable = false))
+ val requiredSchema = StructType(fields)
+
+ val appliedSchema: StructType = SparkFileFormatInternalRowReaderContext.getAppliedRequiredSchema(
Review Comment:
Do we have tests on `SparkFileFormatInternalRowReaderContext` in different Spark versions so that the filters pushed down are different depending on whether the Spark parquet reader can read row index from the parquet files?
##########
hudi-common/src/main/java/org/apache/hudi/common/engine/HoodieReaderContext.java:
##########
@@ -301,9 +311,19 @@ public final UnaryOperator<T> projectRecord(Schema from, Schema to) {
* @return the record position in the base file.
*/
public long extractRecordPosition(T record, Schema schema, String fieldName, long providedPositionIfNeeded) {
+ if (supportsParquetRowIndex()) {
+ Object position = getValue(record, schema, fieldName);
+ if (position != null) {
+ return (long) position;
+ }
Review Comment:
Should we throw an error if the position is not read from the parquet reader when we enable 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631715307
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
Review Comment:
I think the filtering is actually done by batch as well, so I think we wouldn't need to iterate through the rows themselves
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631738403
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
Review Comment:
https://issues.apache.org/jira/browse/HUDI-7693
--
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
Re: [PR] [HUDI-7840] Add position merging to the new file group reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155661535
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
}, {
"hash" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24280",
"triggerID" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"triggerType" : "PUSH"
}, {
"hash" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24281",
"triggerID" : "4d4c0fdc03b72cfb7ad86172a75fcca439e42682",
"triggerType" : "PUSH"
}, {
"hash" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24283",
"triggerID" : "6c15d7a0558284728296d74c9acbc6805230d9a2",
"triggerType" : "PUSH"
}, {
"hash" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0bf72cfded469e3cc1091827cbe4f2f3c16de830",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 6c15d7a0558284728296d74c9acbc6805230d9a2 Azure: [CANCELED](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24283)
* 0bf72cfded469e3cc1091827cbe4f2f3c16de830 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155060689
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264)
* d581b2726ba5047c9e72396820da81ecf1357266 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155498531
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631729256
##########
hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/TestHoodiePositionBasedFileGroupRecordBuffer.java:
##########
@@ -0,0 +1,214 @@
+/*
+ * 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.config.HoodieStorageConfig;
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.engine.HoodieReaderContext;
+import org.apache.hudi.common.model.DeleteRecord;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieRecordMerger;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.TableSchemaResolver;
+import org.apache.hudi.common.table.log.block.HoodieDeleteBlock;
+import org.apache.hudi.common.table.log.block.HoodieLogBlock;
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer;
+import org.apache.hudi.common.table.read.HoodiePositionBasedSchemaHandler;
+import org.apache.hudi.common.table.read.TestHoodieFileGroupReaderOnSpark;
+import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.SchemaTestUtil;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.common.util.collection.ExternalSpillableMap;
+import org.apache.hudi.common.util.collection.Pair;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.exception.HoodieValidationException;
+import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.IndexedRecord;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.hudi.common.engine.HoodieReaderContext.INTERNAL_META_RECORD_KEY;
+import static org.apache.hudi.common.model.WriteOperationType.INSERT;
+import static org.apache.hudi.common.testutils.HoodieTestUtils.createMetaClient;
+import static org.apache.hudi.common.testutils.RawTripTestPayload.recordsToStrings;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestHoodiePositionBasedFileGroupRecordBuffer extends TestHoodieFileGroupReaderOnSpark {
+ private final HoodieTestDataGenerator dataGen = new HoodieTestDataGenerator(0xDEEF);
+ private HoodieTableMetaClient metaClient;
+ private Schema avroSchema;
+ private HoodiePositionBasedFileGroupRecordBuffer<InternalRow> buffer;
+ private String partitionPath;
+
+ public void prepareBuffer(boolean useCustomMerger) throws Exception {
+ Map<String, String> writeConfigs = new HashMap<>();
+ writeConfigs.put(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(), "parquet");
+ writeConfigs.put(KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key(), "_row_key");
+ writeConfigs.put(KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key(), "partition_path");
+ writeConfigs.put("hoodie.datasource.write.precombine.field", "timestamp");
+ writeConfigs.put("hoodie.payload.ordering.field", "timestamp");
+ writeConfigs.put(HoodieTableConfig.HOODIE_TABLE_NAME_KEY, "hoodie_test");
+ writeConfigs.put("hoodie.insert.shuffle.parallelism", "4");
+ writeConfigs.put("hoodie.upsert.shuffle.parallelism", "4");
+ writeConfigs.put("hoodie.bulkinsert.shuffle.parallelism", "2");
+ writeConfigs.put("hoodie.delete.shuffle.parallelism", "1");
+ writeConfigs.put("hoodie.merge.small.file.group.candidates.limit", "0");
+ writeConfigs.put("hoodie.compact.inline", "false");
+ writeConfigs.put(HoodieWriteConfig.WRITE_RECORD_POSITIONS.key(), "true");
+ commitToTable(recordsToStrings(dataGen.generateInserts("001", 100)), INSERT.value(), writeConfigs);
+
+ String[] partitionPaths = dataGen.getPartitionPaths();
+ String[] partitionValues = new String[1];
+ partitionPath = partitionPaths[0];
+ partitionValues[0] = partitionPath;
+
+ metaClient = createMetaClient(getStorageConf(), getBasePath());
+ avroSchema = new TableSchemaResolver(metaClient).getTableAvroSchema();
+ Option<String[]> partitionFields = metaClient.getTableConfig().getPartitionFields();
+ Option<String> partitionNameOpt = StringUtils.isNullOrEmpty(partitionPaths[0])
+ ? Option.empty() : Option.of(partitionPaths[0]);
+
+ HoodieReaderContext ctx = getHoodieReaderContext(getBasePath(), avroSchema, getStorageConf());
+ ctx.setTablePath(metaClient.getBasePathV2().toString());
+ ctx.setLatestCommitTime(metaClient.createNewInstantTime());
+ ctx.setShouldMergeUseRecordPosition(true);
+ ctx.setHasBootstrapBaseFile(false);
+ ctx.setHasLogFiles(true);
+ ctx.setNeedsBootstrapMerge(false);
Review Comment:
So `hasBootstrapBaseFile` as true and `needsBootstrapMerge` as true are covered by other tests, 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631742303
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
Review Comment:
So iterating through the rows are still needed for stitching? The filtering may still happen within the parquet page/batch since the page level filtering is based on the column stats, if that is what you're talking about.
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155599995
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
}, {
"hash" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275",
"triggerID" : "d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1",
"triggerType" : "PUSH"
}, {
"hash" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1ce1d753818efb6be00c20fb5a8dd141c7c47f00",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* d00f2862fb8dd8a84fcc5aa1900e76577b8a9bf1 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24275)
* 1ce1d753818efb6be00c20fb5a8dd141c7c47f00 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155074344
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264)
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4e6335c7cfb18881776d572954558a41aa33b91d 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2153925430
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2153778377
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #11413:
URL: https://github.com/apache/hudi/pull/11413#issuecomment-2155087634
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264",
"triggerID" : "1a1ca64bec2fb94acce596934dd636b77cb0aca7",
"triggerType" : "PUSH"
}, {
"hash" : "d581b2726ba5047c9e72396820da81ecf1357266",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d581b2726ba5047c9e72396820da81ecf1357266",
"triggerType" : "PUSH"
}, {
"hash" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273",
"triggerID" : "4e6335c7cfb18881776d572954558a41aa33b91d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1a1ca64bec2fb94acce596934dd636b77cb0aca7 Azure: [SUCCESS](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24264)
* d581b2726ba5047c9e72396820da81ecf1357266 UNKNOWN
* 4e6335c7cfb18881776d572954558a41aa33b91d Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=24273)
<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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631710563
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
+ //we need to iterate row by row when merging
+ new CachingIterator[InternalRow] {
+ val combinedRow = new JoinedRow()
+
+ //position column will always be at the end of the row
+ private def getPos(row: InternalRow): Long = {
+ row.getLong(row.numFields-1)
+ }
+
+ private def getNextSkeleton: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ private def getNextData: (InternalRow, Long) = {
+ val nextDataRow = dataFileIterator.next().asInstanceOf[InternalRow]
+ (nextDataRow, getPos(nextDataRow))
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+
+ override protected def doHasNext(): Boolean = {
+ if (!dataFileIterator.hasNext || !skeletonFileIterator.hasNext) {
+ false
+ } else {
+ var nextSkeleton = getNextSkeleton
+ var nextData = getNextData
+ while (nextSkeleton._2 != nextData._2) {
+ if (nextSkeleton._2 > nextData._2) {
+ if (!dataFileIterator.hasNext) {
+ return false
+ } else {
+ nextData = getNextData
+ }
} else {
- vecs(i) = d.column(i - s.numCols())
+ if (!skeletonFileIterator.hasNext) {
+ return false
+ } else {
+ nextSkeleton = getNextSkeleton
+ }
}
}
- assert(s.numRows() == d.numRows())
- sparkAdapter.makeColumnarBatch(vecs, s.numRows())
- case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ nextRecord = combinedRow(skeletonProjection.apply(nextSkeleton._1), dataProjection.apply(nextData._1))
+ true
+ }
}
}
+ } else {
+ new ClosableIterator[Any] {
+ val combinedRow = new JoinedRow()
- override def close(): Unit = {
- skeletonFileIterator.close()
- dataFileIterator.close()
- }
- }.asInstanceOf[ClosableIterator[InternalRow]]
+ override def hasNext: Boolean = {
+ //If the iterators are out of sync it is probably due to filter pushdown
+ checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
+ "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
+ dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ }
+
+ override def next(): Any = {
+ (skeletonFileIterator.next(), dataFileIterator.next()) match {
+ case (s: ColumnarBatch, d: ColumnarBatch) =>
+ //This will not be used until [HUDI-7693] is implemented
+ val numCols = s.numCols() + d.numCols()
+ val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
+ for (i <- 0 until numCols) {
+ if (i < s.numCols()) {
+ vecs(i) = s.column(i)
+ } else {
+ vecs(i) = d.column(i - s.numCols())
+ }
+ }
+ assert(s.numRows() == d.numRows())
+ sparkAdapter.makeColumnarBatch(vecs, s.numRows())
+ case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
+ case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ }
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+ }.asInstanceOf[ClosableIterator[InternalRow]]
+ }
}
}
+
+object SparkFileFormatInternalRowReaderContext {
+ // From "namedExpressions.scala": Used to construct to record position field metadata.
+ private val FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY = "__file_source_generated_metadata_col"
+ private val FILE_SOURCE_METADATA_COL_ATTR_KEY = "__file_source_metadata_col"
+ private val METADATA_COL_ATTR_KEY = "__metadata_col"
+
+ def getAppliedRequiredSchema(requiredSchema: StructType, shouldAddRecordPosition: Boolean): StructType = {
+ if (shouldAddRecordPosition) {
+ val metadata = new MetadataBuilder()
+ .putString(METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .putBoolean(FILE_SOURCE_METADATA_COL_ATTR_KEY, value = true)
+ .putString(FILE_SOURCE_GENERATED_METADATA_COL_ATTR_KEY, ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ .build()
+ val rowIndexField = StructField(ROW_INDEX_TEMPORARY_COLUMN_NAME, LongType, nullable = false, metadata)
+ StructType(requiredSchema.fields.filterNot(isIndexTempColumn) :+ rowIndexField)
+ } else {
+ requiredSchema
+ }
+ }
+
+ /**
+ * Only valid if there is support for record positions and no log files
Review Comment:
yes
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631715749
##########
hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala:
##########
@@ -116,45 +143,154 @@ class SparkFileFormatInternalRowReaderContext(parquetFileReader: SparkParquetRea
skeletonRequiredSchema: Schema,
dataFileIterator: ClosableIterator[InternalRow],
dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
- doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]],
- dataFileIterator.asInstanceOf[ClosableIterator[Any]])
+ doBootstrapMerge(skeletonFileIterator.asInstanceOf[ClosableIterator[Any]], skeletonRequiredSchema,
+ dataFileIterator.asInstanceOf[ClosableIterator[Any]], dataRequiredSchema)
}
- protected def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any], dataFileIterator: ClosableIterator[Any]): ClosableIterator[InternalRow] = {
- new ClosableIterator[Any] {
- val combinedRow = new JoinedRow()
+ private def doBootstrapMerge(skeletonFileIterator: ClosableIterator[Any],
+ skeletonRequiredSchema: Schema,
+ dataFileIterator: ClosableIterator[Any],
+ dataRequiredSchema: Schema): ClosableIterator[InternalRow] = {
+ if (supportsPositionField()) {
+ assert(AvroSchemaUtils.containsFieldInSchema(skeletonRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ assert(AvroSchemaUtils.containsFieldInSchema(dataRequiredSchema, ROW_INDEX_TEMPORARY_COLUMN_NAME))
+ val rowIndexColumn = new java.util.HashSet[String]()
+ rowIndexColumn.add(ROW_INDEX_TEMPORARY_COLUMN_NAME)
+ //always remove the row index column from the skeleton because the data file will also have the same column
+ val skeletonProjection = projectRecord(skeletonRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(skeletonRequiredSchema, rowIndexColumn))
- override def hasNext: Boolean = {
- //If the iterators are out of sync it is probably due to filter pushdown
- checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
- "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
- dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ //If we need to do position based merging with log files we will leave the row index column at the end
+ val dataProjection = if (getHasLogFiles && getUseRecordPosition) {
+ getIdentityProjection
+ } else {
+ projectRecord(dataRequiredSchema,
+ AvroSchemaUtils.removeFieldsFromSchema(dataRequiredSchema, rowIndexColumn))
}
- override def next(): Any = {
- (skeletonFileIterator.next(), dataFileIterator.next()) match {
- case (s: ColumnarBatch, d: ColumnarBatch) =>
- val numCols = s.numCols() + d.numCols()
- val vecs: Array[ColumnVector] = new Array[ColumnVector](numCols)
- for (i <- 0 until numCols) {
- if (i < s.numCols()) {
- vecs(i) = s.column(i)
+ //Always use internal row for positional merge because
+ //we need to iterate row by row when merging
+ new CachingIterator[InternalRow] {
+ val combinedRow = new JoinedRow()
+
+ //position column will always be at the end of the row
+ private def getPos(row: InternalRow): Long = {
+ row.getLong(row.numFields-1)
+ }
+
+ private def getNextSkeleton: (InternalRow, Long) = {
+ val nextSkeletonRow = skeletonFileIterator.next().asInstanceOf[InternalRow]
+ (nextSkeletonRow, getPos(nextSkeletonRow))
+ }
+
+ private def getNextData: (InternalRow, Long) = {
+ val nextDataRow = dataFileIterator.next().asInstanceOf[InternalRow]
+ (nextDataRow, getPos(nextDataRow))
+ }
+
+ override def close(): Unit = {
+ skeletonFileIterator.close()
+ dataFileIterator.close()
+ }
+
+ override protected def doHasNext(): Boolean = {
+ if (!dataFileIterator.hasNext || !skeletonFileIterator.hasNext) {
+ false
+ } else {
+ var nextSkeleton = getNextSkeleton
+ var nextData = getNextData
+ while (nextSkeleton._2 != nextData._2) {
+ if (nextSkeleton._2 > nextData._2) {
+ if (!dataFileIterator.hasNext) {
+ return false
+ } else {
+ nextData = getNextData
+ }
} else {
- vecs(i) = d.column(i - s.numCols())
+ if (!skeletonFileIterator.hasNext) {
+ return false
+ } else {
+ nextSkeleton = getNextSkeleton
+ }
}
}
- assert(s.numRows() == d.numRows())
- sparkAdapter.makeColumnarBatch(vecs, s.numRows())
- case (_: ColumnarBatch, _: InternalRow) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (_: InternalRow, _: ColumnarBatch) => throw new IllegalStateException("InternalRow ColumnVector mismatch")
- case (s: InternalRow, d: InternalRow) => combinedRow(s, d)
+ nextRecord = combinedRow(skeletonProjection.apply(nextSkeleton._1), dataProjection.apply(nextData._1))
+ true
+ }
}
}
+ } else {
+ new ClosableIterator[Any] {
+ val combinedRow = new JoinedRow()
- override def close(): Unit = {
- skeletonFileIterator.close()
- dataFileIterator.close()
- }
- }.asInstanceOf[ClosableIterator[InternalRow]]
+ override def hasNext: Boolean = {
+ //If the iterators are out of sync it is probably due to filter pushdown
+ checkState(dataFileIterator.hasNext == skeletonFileIterator.hasNext,
+ "Bootstrap data-file iterator and skeleton-file iterator have to be in-sync!")
+ dataFileIterator.hasNext && skeletonFileIterator.hasNext
+ }
+
+ override def next(): Any = {
+ (skeletonFileIterator.next(), dataFileIterator.next()) match {
Review Comment:
Yes, that hasn't changed
--
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
Re: [PR] [HUDI-7840] Add position merging to fg reader [hudi]
Posted by "jonvex (via GitHub)" <gi...@apache.org>.
jonvex commented on code in PR #11413:
URL: https://github.com/apache/hudi/pull/11413#discussion_r1631723794
##########
hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/execution/datasources/parquet/TestSparkFileFormatInternalRowReaderContext.scala:
##########
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.execution.datasources.parquet
+
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext
+import org.apache.hudi.SparkFileFormatInternalRowReaderContext.filterIsSafeForBootstrap
+import org.apache.hudi.common.model.HoodieRecord
+import org.apache.hudi.common.table.read.HoodiePositionBasedFileGroupRecordBuffer.ROW_INDEX_TEMPORARY_COLUMN_NAME
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness
+import org.apache.spark.sql.sources.{And, IsNotNull, Or}
+import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertTrue}
+import org.junit.jupiter.api.Test
+
+class TestSparkFileFormatInternalRowReaderContext extends SparkClientFunctionalTestHarness {
+
+ @Test
+ def testBootstrapFilters(): Unit = {
+ val recordKeyField = HoodieRecord.HoodieMetadataField.RECORD_KEY_METADATA_FIELD.getFieldName
+ val commitTimeField = HoodieRecord.HoodieMetadataField.COMMIT_TIME_METADATA_FIELD.getFieldName
+
+ val recordKeyFilter = IsNotNull(recordKeyField)
+ assertTrue(filterIsSafeForBootstrap(recordKeyFilter))
+ val commitTimeFilter = IsNotNull(commitTimeField)
+ assertTrue(filterIsSafeForBootstrap(commitTimeFilter))
+
+ val dataFieldFilter = IsNotNull("someotherfield")
+ assertTrue(filterIsSafeForBootstrap(dataFieldFilter))
+
+ val legalComplexFilter = Or(recordKeyFilter, commitTimeFilter)
+ assertTrue(filterIsSafeForBootstrap(legalComplexFilter))
+
+ val illegalComplexFilter = Or(recordKeyFilter, dataFieldFilter)
+ assertFalse(filterIsSafeForBootstrap(illegalComplexFilter))
+
+ val illegalNestedFilter = And(legalComplexFilter, illegalComplexFilter)
+ assertFalse(filterIsSafeForBootstrap(illegalNestedFilter))
+
+ val legalNestedFilter = And(legalComplexFilter, recordKeyFilter)
+ assertTrue(filterIsSafeForBootstrap(legalNestedFilter))
+ }
+
+ @Test
+ def testGetAppliedRequiredSchema(): Unit = {
+ val fields = Array(
+ StructField("column_a", LongType, nullable = false),
+ StructField("column_b", StringType, nullable = false))
+ val requiredSchema = StructType(fields)
+
+ val appliedSchema: StructType = SparkFileFormatInternalRowReaderContext.getAppliedRequiredSchema(
Review Comment:
TestFiltersInFileGroupReader does tests to ensure that filters are pushed down when they should be. I also set breakpoints to make sure the filtering was actually happening.
--
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