You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by GitBox <gi...@apache.org> on 2022/12/11 15:02:56 UTC
[GitHub] [hudi] loukey-lj opened a new pull request, #7429: Record Level Index
loukey-lj opened a new pull request, #7429:
URL: https://github.com/apache/hudi/pull/7429
### Change Logs
_Describe context and summary for this change. Highlight if any code was copied._
### Impact
_Describe any public API or user-facing feature change or any performance impact._
### Risk level (write none, low medium or high below)
_If medium or high, explain what verification was done to mitigate the risks._
### Documentation Update
_Describe any necessary documentation update if there is any new feature, config, or user-facing change_
- _The config description must be updated if new configs are added or the default value of the configs are changed_
- _Any new feature or user-facing change requires updating the Hudi website. Please create a Jira ticket, attach the
ticket number here and follow the [instruction](https://hudi.apache.org/contribute/developer-setup#website) to make
changes to the website._
### 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
[GitHub] [hudi] hudi-bot commented on pull request #7429: Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345620282
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345842670
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
}, {
"hash" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"triggerType" : "PUSH"
}, {
"hash" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13631",
"triggerID" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626)
* e37581e0d9086e050b552f1219ebce0520b68f8c UNKNOWN
* dfa17c8e84b8e8f66c867e504aa9c434e1e2c255 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13631)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345976825
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
}, {
"hash" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"triggerType" : "PUSH"
}, {
"hash" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13631",
"triggerID" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* e37581e0d9086e050b552f1219ebce0520b68f8c UNKNOWN
* dfa17c8e84b8e8f66c867e504aa9c434e1e2c255 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13631)
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] loukey-lj commented on pull request #7429: [HUDI-53] Record Level Index
Posted by "loukey-lj (via GitHub)" <gi...@apache.org>.
loukey-lj commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1495646165
@nsivabalan Too disappointed, after half a year, the community has not made any progress
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345836945
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
}, {
"hash" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"triggerType" : "PUSH"
}, {
"hash" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dfa17c8e84b8e8f66c867e504aa9c434e1e2c255",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626)
* e37581e0d9086e050b552f1219ebce0520b68f8c UNKNOWN
* dfa17c8e84b8e8f66c867e504aa9c434e1e2c255 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] loukey-lj commented on a diff in pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
loukey-lj commented on code in PR #7429:
URL: https://github.com/apache/hudi/pull/7429#discussion_r1058019890
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -967,6 +974,96 @@ public static HoodieData<HoodieRecord> convertFilesToColumnStatsRecords(HoodieEn
return allRecordsRDD;
}
+ /**
+ * Convert added and deleted action metadata to record level index index records.
+ */
+ public static HoodieData<HoodieRecord> convertFilesToRecordLevelIndex(HoodieEngineContext engineContext,
+ Map<String, List<String>> partitionToDeletedFiles,
+ Map<String, Map<String, Long>> partitionToAppendedFiles,
+ MetadataRecordsGenerationParams recordsGenerationParams,
+ String instantTime,
+ String datasetBasePath,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ HoodieData<HoodieRecord> allRecordsRDD = engineContext.emptyHoodieData();
+
+ List<Pair<String, List<String>>> partitionToDeletedFilesList = partitionToDeletedFiles.entrySet()
+ .stream().map(e -> Pair.of(e.getKey(), e.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToDelete = partitionToDeletedFilesList.stream().flatMap(e -> e.getRight().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ int parallelism = Math.max(Math.min(filesToDelete.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToDeletedFilesRDD = engineContext.parallelize(filesToDelete, parallelism);
+
+ HoodieData<HoodieRecord> deletedFilesRecordsRDD = partitionToDeletedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, true, keyGeneratorOpt);
+ });
+ allRecordsRDD = allRecordsRDD.union(deletedFilesRecordsRDD);
+
+ List<Pair<String, Map<String, Long>>> partitionToAppendedFilesList = partitionToAppendedFiles.entrySet()
+ .stream().map(entry -> Pair.of(entry.getKey(), entry.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToAppended = partitionToAppendedFilesList.stream().flatMap(e -> e.getRight().keySet().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ parallelism = Math.max(Math.min(filesToAppended.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToAppendedFilesRDD = engineContext.parallelize(filesToAppended, parallelism);
+
+ HoodieData<HoodieRecord> appendedFilesRecordsRDD = partitionToAppendedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, false, keyGeneratorOpt);
+ });
+
+ allRecordsRDD = allRecordsRDD.union(appendedFilesRecordsRDD);
+
+ return allRecordsRDD;
+ }
+
+ private static Iterator<HoodieRecord> getRecordIndexFromParquetFile(
+ Configuration conf,
+ String datasetBasePath,
+ Pair<String, String> partitionToFilePair,
+ boolean isDeleted,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ final String partition = getPartitionIdentifier(partitionToFilePair.getLeft());
+ final String fileName = partitionToFilePair.getRight();
+ final String fileId = FSUtils.getFileId(fileName);
+ final String fileCommitTime = FSUtils.getCommitTime(fileName);
+
+ Path dataFilePath = new Path(datasetBasePath, String.format("%s%s%s", partition, Path.SEPARATOR, fileName));
+
+ BaseFileUtils baseFileUtils = BaseFileUtils.getInstance(dataFilePath.toString());
+ Iterator<HoodieKey> recordKeyIterator;
+ if (keyGeneratorOpt.isPresent()) {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath, keyGeneratorOpt);
+ } else {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath);
+ }
+ // final List<Long> blockRecordSize = baseFileUtils.getBlockRecordSize(conf, dataFilePath);
+
+ return new Iterator<HoodieRecord>() {
+ long size = 0L;
+
+ @Override
+ public boolean hasNext() {
+ return recordKeyIterator.hasNext();
+ }
+
+ @Override
+ public HoodieRecord next() {
+ size++;
+ // int rowGroupIndex = findRowGroupIndex();
+ int rowGroupIndex = -1;
+ HoodieKey next = recordKeyIterator.next();
+ return HoodieMetadataPayload.createRecordLevelIndexRecord(next.getRecordKey(), next.getPartitionPath(), fileId, rowGroupIndex, isDeleted, fileCommitTime, HoodieOperation.INSERT);
+ }
+
+ // int findRowGroupIndex() {
Review Comment:
RowGroupIndex a reserved attribute, which indicates the serial number of the rowGroup in the parquet file, and it has not been collected 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
[GitHub] [hudi] hudi-bot commented on pull request #7429: Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345583533
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 UNKNOWN
<details>
<summary>Bot commands</summary>
@hudi-bot supports the following commands:
- `@hudi-bot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345832315
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
}, {
"hash" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "e37581e0d9086e050b552f1219ebce0520b68f8c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 Azure: [FAILURE](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626)
* e37581e0d9086e050b552f1219ebce0520b68f8c 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-53] Record Level Index [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1986746064
Closing this as the Record Level Index is landed in #8758 and included in Hudi 0.14.0 release.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] loukey-lj commented on a diff in pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
loukey-lj commented on code in PR #7429:
URL: https://github.com/apache/hudi/pull/7429#discussion_r1058017034
##########
hudi-common/src/main/avro/HoodieMetadata.avsc:
##########
@@ -358,6 +358,45 @@
}
],
"default" : null
+ },
+ {
+ "name": "recordLevelIndexMetadata",
+ "doc": "Contains information about record keys and their location in the dataset",
+ "type": [
+ "null",
+ {
+ "type": "record",
+ "name": "HoodieMetadataRecordLevelIndex",
+ "fields": [
+ {
+ "name": "partition",
+ "type": "string",
+ "doc": "Partition which contains the record",
+ "avro.java.string": "String"
+ },
+ {
+ "name": "fileId",
+ "type": "string",
+ "doc": "FileId of file group"
+ },
+ {
+ "name": "rowGroupIndex",
Review Comment:
Thanks for review.
This is a parameter reserved for parquet partial updates
https://github.com/apache/hudi/pull/6612
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] nsivabalan commented on pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1361535138
hey, thanks for the patch. we do already have a patch on this end https://github.com/apache/hudi/pull/5581 just that it was based of of 0.10.1. And it has some enhancements to foundational metadata table before we can bring in record level index. Give me 2 days. I am half way through the other patch to understand how we can split it into multiple patches and for actual record level index, we can incorporate from this patch may be. I will keep you posted.
will share the concrete plans here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] hudi-bot commented on pull request #7429: Record Level Index
Posted by GitBox <gi...@apache.org>.
hudi-bot commented on PR #7429:
URL: https://github.com/apache/hudi/pull/7429#issuecomment-1345585006
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626",
"triggerID" : "8fa28be0d20f144d9000c672e9500e42a5a66055",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8fa28be0d20f144d9000c672e9500e42a5a66055 Azure: [PENDING](https://dev.azure.com/apache-hudi-ci-org/785b6ef4-2f42-4a89-8f0e-5f0d7039a0cc/_build/results?buildId=13626)
<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-53] Record Level Index [hudi]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua closed pull request #7429: [HUDI-53] Record Level Index
URL: https://github.com/apache/hudi/pull/7429
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [hudi] loukey-lj commented on a diff in pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
loukey-lj commented on code in PR #7429:
URL: https://github.com/apache/hudi/pull/7429#discussion_r1058019890
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -967,6 +974,96 @@ public static HoodieData<HoodieRecord> convertFilesToColumnStatsRecords(HoodieEn
return allRecordsRDD;
}
+ /**
+ * Convert added and deleted action metadata to record level index index records.
+ */
+ public static HoodieData<HoodieRecord> convertFilesToRecordLevelIndex(HoodieEngineContext engineContext,
+ Map<String, List<String>> partitionToDeletedFiles,
+ Map<String, Map<String, Long>> partitionToAppendedFiles,
+ MetadataRecordsGenerationParams recordsGenerationParams,
+ String instantTime,
+ String datasetBasePath,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ HoodieData<HoodieRecord> allRecordsRDD = engineContext.emptyHoodieData();
+
+ List<Pair<String, List<String>>> partitionToDeletedFilesList = partitionToDeletedFiles.entrySet()
+ .stream().map(e -> Pair.of(e.getKey(), e.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToDelete = partitionToDeletedFilesList.stream().flatMap(e -> e.getRight().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ int parallelism = Math.max(Math.min(filesToDelete.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToDeletedFilesRDD = engineContext.parallelize(filesToDelete, parallelism);
+
+ HoodieData<HoodieRecord> deletedFilesRecordsRDD = partitionToDeletedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, true, keyGeneratorOpt);
+ });
+ allRecordsRDD = allRecordsRDD.union(deletedFilesRecordsRDD);
+
+ List<Pair<String, Map<String, Long>>> partitionToAppendedFilesList = partitionToAppendedFiles.entrySet()
+ .stream().map(entry -> Pair.of(entry.getKey(), entry.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToAppended = partitionToAppendedFilesList.stream().flatMap(e -> e.getRight().keySet().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ parallelism = Math.max(Math.min(filesToAppended.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToAppendedFilesRDD = engineContext.parallelize(filesToAppended, parallelism);
+
+ HoodieData<HoodieRecord> appendedFilesRecordsRDD = partitionToAppendedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, false, keyGeneratorOpt);
+ });
+
+ allRecordsRDD = allRecordsRDD.union(appendedFilesRecordsRDD);
+
+ return allRecordsRDD;
+ }
+
+ private static Iterator<HoodieRecord> getRecordIndexFromParquetFile(
+ Configuration conf,
+ String datasetBasePath,
+ Pair<String, String> partitionToFilePair,
+ boolean isDeleted,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ final String partition = getPartitionIdentifier(partitionToFilePair.getLeft());
+ final String fileName = partitionToFilePair.getRight();
+ final String fileId = FSUtils.getFileId(fileName);
+ final String fileCommitTime = FSUtils.getCommitTime(fileName);
+
+ Path dataFilePath = new Path(datasetBasePath, String.format("%s%s%s", partition, Path.SEPARATOR, fileName));
+
+ BaseFileUtils baseFileUtils = BaseFileUtils.getInstance(dataFilePath.toString());
+ Iterator<HoodieKey> recordKeyIterator;
+ if (keyGeneratorOpt.isPresent()) {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath, keyGeneratorOpt);
+ } else {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath);
+ }
+ // final List<Long> blockRecordSize = baseFileUtils.getBlockRecordSize(conf, dataFilePath);
+
+ return new Iterator<HoodieRecord>() {
+ long size = 0L;
+
+ @Override
+ public boolean hasNext() {
+ return recordKeyIterator.hasNext();
+ }
+
+ @Override
+ public HoodieRecord next() {
+ size++;
+ // int rowGroupIndex = findRowGroupIndex();
+ int rowGroupIndex = -1;
+ HoodieKey next = recordKeyIterator.next();
+ return HoodieMetadataPayload.createRecordLevelIndexRecord(next.getRecordKey(), next.getPartitionPath(), fileId, rowGroupIndex, isDeleted, fileCommitTime, HoodieOperation.INSERT);
+ }
+
+ // int findRowGroupIndex() {
Review Comment:
RowGroupIndex is a reserved attribute, which indicates the serial number of the rowGroup in the parquet file, and it has not been collected 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
[GitHub] [hudi] nsivabalan commented on a diff in pull request #7429: [HUDI-53] Record Level Index
Posted by GitBox <gi...@apache.org>.
nsivabalan commented on code in PR #7429:
URL: https://github.com/apache/hudi/pull/7429#discussion_r1054551194
##########
hudi-common/src/main/java/org/apache/hudi/metadata/HoodieTableMetadataUtil.java:
##########
@@ -967,6 +974,96 @@ public static HoodieData<HoodieRecord> convertFilesToColumnStatsRecords(HoodieEn
return allRecordsRDD;
}
+ /**
+ * Convert added and deleted action metadata to record level index index records.
+ */
+ public static HoodieData<HoodieRecord> convertFilesToRecordLevelIndex(HoodieEngineContext engineContext,
+ Map<String, List<String>> partitionToDeletedFiles,
+ Map<String, Map<String, Long>> partitionToAppendedFiles,
+ MetadataRecordsGenerationParams recordsGenerationParams,
+ String instantTime,
+ String datasetBasePath,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ HoodieData<HoodieRecord> allRecordsRDD = engineContext.emptyHoodieData();
+
+ List<Pair<String, List<String>>> partitionToDeletedFilesList = partitionToDeletedFiles.entrySet()
+ .stream().map(e -> Pair.of(e.getKey(), e.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToDelete = partitionToDeletedFilesList.stream().flatMap(e -> e.getRight().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ int parallelism = Math.max(Math.min(filesToDelete.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToDeletedFilesRDD = engineContext.parallelize(filesToDelete, parallelism);
+
+ HoodieData<HoodieRecord> deletedFilesRecordsRDD = partitionToDeletedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, true, keyGeneratorOpt);
+ });
+ allRecordsRDD = allRecordsRDD.union(deletedFilesRecordsRDD);
+
+ List<Pair<String, Map<String, Long>>> partitionToAppendedFilesList = partitionToAppendedFiles.entrySet()
+ .stream().map(entry -> Pair.of(entry.getKey(), entry.getValue())).collect(Collectors.toList());
+ List<Pair<String, String>> filesToAppended = partitionToAppendedFilesList.stream().flatMap(e -> e.getRight().keySet().stream().map(f -> Pair.of(e.getLeft(), f))).collect(Collectors.toList());
+ parallelism = Math.max(Math.min(filesToAppended.size(), recordsGenerationParams.getRecordLevelIndexParallelism()), 1);
+ HoodieData<Pair<String, String>> partitionToAppendedFilesRDD = engineContext.parallelize(filesToAppended, parallelism);
+
+ HoodieData<HoodieRecord> appendedFilesRecordsRDD = partitionToAppendedFilesRDD.filter(p -> FSUtils.isBaseFile(new Path(p.getRight()))).flatMap(partitionToDeletedFilePair -> {
+ return getRecordIndexFromParquetFile(new Configuration(), datasetBasePath, partitionToDeletedFilePair, false, keyGeneratorOpt);
+ });
+
+ allRecordsRDD = allRecordsRDD.union(appendedFilesRecordsRDD);
+
+ return allRecordsRDD;
+ }
+
+ private static Iterator<HoodieRecord> getRecordIndexFromParquetFile(
+ Configuration conf,
+ String datasetBasePath,
+ Pair<String, String> partitionToFilePair,
+ boolean isDeleted,
+ Option<BaseKeyGenerator> keyGeneratorOpt) {
+ final String partition = getPartitionIdentifier(partitionToFilePair.getLeft());
+ final String fileName = partitionToFilePair.getRight();
+ final String fileId = FSUtils.getFileId(fileName);
+ final String fileCommitTime = FSUtils.getCommitTime(fileName);
+
+ Path dataFilePath = new Path(datasetBasePath, String.format("%s%s%s", partition, Path.SEPARATOR, fileName));
+
+ BaseFileUtils baseFileUtils = BaseFileUtils.getInstance(dataFilePath.toString());
+ Iterator<HoodieKey> recordKeyIterator;
+ if (keyGeneratorOpt.isPresent()) {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath, keyGeneratorOpt);
+ } else {
+ recordKeyIterator = baseFileUtils.getHoodieKeyIterator(conf, dataFilePath);
+ }
+ // final List<Long> blockRecordSize = baseFileUtils.getBlockRecordSize(conf, dataFilePath);
+
+ return new Iterator<HoodieRecord>() {
+ long size = 0L;
+
+ @Override
+ public boolean hasNext() {
+ return recordKeyIterator.hasNext();
+ }
+
+ @Override
+ public HoodieRecord next() {
+ size++;
+ // int rowGroupIndex = findRowGroupIndex();
+ int rowGroupIndex = -1;
+ HoodieKey next = recordKeyIterator.next();
+ return HoodieMetadataPayload.createRecordLevelIndexRecord(next.getRecordKey(), next.getPartitionPath(), fileId, rowGroupIndex, isDeleted, fileCommitTime, HoodieOperation.INSERT);
+ }
+
+ // int findRowGroupIndex() {
Review Comment:
guess its not yet used as of this patch. but w/ hfile, may I know how does this row group index help us?
##########
hudi-common/src/main/avro/HoodieMetadata.avsc:
##########
@@ -358,6 +358,45 @@
}
],
"default" : null
+ },
+ {
+ "name": "recordLevelIndexMetadata",
+ "doc": "Contains information about record keys and their location in the dataset",
+ "type": [
+ "null",
+ {
+ "type": "record",
+ "name": "HoodieMetadataRecordLevelIndex",
+ "fields": [
+ {
+ "name": "partition",
+ "type": "string",
+ "doc": "Partition which contains the record",
+ "avro.java.string": "String"
+ },
+ {
+ "name": "fileId",
+ "type": "string",
+ "doc": "FileId of file group"
+ },
+ {
+ "name": "rowGroupIndex",
Review Comment:
may I know whats the purpose of row group index ?
--
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