You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "voonhous (via GitHub)" <gi...@apache.org> on 2026/07/14 15:24:33 UTC
[PR] 19249 resolve required columns from table schema [hudi]
voonhous opened a new pull request, #19288:
URL: https://github.com/apache/hudi/pull/19288
### Describe the issue this Pull Request addresses
<!-- Either describe the issue inline here with motivation behind the changes
(or) link to an issue by including `Closes #<issue-number>` for context.
If this PR includes changes to the storage format, public APIs,
or has breaking changes, use `!` (e.g., feat!: ...) -->
### Summary and Changelog
<!-- Short, plain-English summary of what users gain or what changed in behavior.
Followed by a detailed log of all the changes. Highlight if any code was copied. -->
### Impact
<!-- Describe any public API or user-facing feature change or any performance impact. -->
### Risk Level
<!-- Accepted values: none, low, medium or high. Other than `none`, explain the risk.
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. If not, put "none".
- 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 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)
- [ ] Enough context is provided in the sections above
- [ ] Adequate tests were added if applicable
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3663203416
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +624,94 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} keeps only metastore data
+ * columns, and the merge read path recovers any name the metastore does not carry from the resolved
+ * table schema ({@link #appendMissingMergeRequiredColumns}). For those two fields that mirrors the
+ * file-group reader's own schema gate, so tables whose schema has neither read nothing extra; the
+ * ordering, record-key and delete-key names the reader adds without a schema check, but they are
+ * table columns whenever the table config is coherent.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
+ requiredColumnNames.add(OPERATION_METADATA_FIELD);
+ String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
+ String deleteMarker = tableConfig.getProps().getProperty(DELETE_MARKER);
+ // DeleteContext only honors a custom delete key when the marker value is also set
+ if (!StringUtils.isNullOrEmpty(deleteKey) && !StringUtils.isNullOrEmpty(deleteMarker)) {
+ requiredColumnNames.add(deleteKey);
+ }
+ return requiredColumnNames;
+ }
+
+ /**
+ * Returns {@code projection} extended with handles for the merge-required column names it does not
+ * already carry (matched case-insensitively) and the metastore could not resolve, typed from
+ * {@code dataSchema} via {@link #toColumnHandle}: the table-config-derived names
+ * ({@link #mergeRequiredColumnNames}) plus, under a CUSTOM merge mode, the mandatory fields declared
+ * by the same resolved merger the file-group reader will use. Names absent from the schema as well are
+ * dropped -- for {@code _hoodie_is_deleted} and {@code _hoodie_operation} that mirrors the file-group
+ * reader's own schema gate; the rest are table columns whenever the table config is coherent. A
+ * metastore that omits a field the table schema carries -- hive sync with
+ * {@code omit_metadata_fields=true} drops {@code _hoodie_operation}, hand-written external DDL can drop
+ * any data column -- must not starve the base-file read of it; the
+ * {@code HudiTrinoReaderContext.getFileRecordIterator} guard would otherwise fail the read. Called on
+ * the merge read path only, where the table schema is already resolved, so the recovery costs no I/O.
+ */
+ public static List<HiveColumnHandle> appendMissingMergeRequiredColumns(
+ HoodieSchema dataSchema,
+ List<HiveColumnHandle> projection,
+ HoodieTableConfig tableConfig,
+ TypedProperties readerProps)
+ {
+ Pair<RecordMergeMode, String> mergeModeAndStrategyId = resolveMergeModeAndStrategyId(tableConfig);
+ LinkedHashSet<String> requiredNames = mergeRequiredColumnNames(tableConfig, mergeModeAndStrategyId.getLeft());
+ // A CUSTOM merger can declare mandatory fields of its own (getMandatoryFieldsForMerging); ask the
+ // same resolved merger the file-group reader will use, against the same resolved schema, so its
+ // declarations are recovered too. Resolution is classloading only -- no I/O.
+ if (mergeModeAndStrategyId.getLeft() == RecordMergeMode.CUSTOM && !StringUtils.isNullOrEmpty(mergeModeAndStrategyId.getRight())) {
+ String mergeImplClasses = readerProps.getString(RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY,
+ readerProps.getString(RECORD_MERGE_IMPL_CLASSES_DEPRECATED_WRITE_CONFIG_KEY, ""));
+ Option<HoodieRecordMerger> merger = HoodieRecordUtils.createValidRecordMerger(EngineType.JAVA, mergeImplClasses, mergeModeAndStrategyId.getRight());
+ if (merger.isPresent()) {
Review Comment:
The empty-strategy-id and unresolvable-merger guards this branch adds have no test, while the same two cases are pinned for `usesNonProjectionCompatibleMerger` in `TestHudiUtilColumnHandles`. Worth mirroring those two assertions onto `appendMissingMergeRequiredColumns`, so removing either guard fails a test instead of turning the `HUDI_BAD_DATA` path back into an NPE.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5106325312
@wombatu-kun Can i request for another round of review please? For the comments addressed and if you have no further comments, can you please help to "resolve" them so they are collapsed?
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5090770476
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 74e1c977c82f3f88867bd5fdc9774d4366592e57 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566)
* 6569ea38b806659ffd0e192334fa2f4dd4609efc Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5027060306
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 50cf40647d30e797753d6b2815adfdfa4b3aedec Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483)
* 74e1c977c82f3f88867bd5fdc9774d4366592e57 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5091942192
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6569ea38b806659ffd0e192334fa2f4dd4609efc Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733)
* 65f09d7f048606cb5f625bdec33492c7362effda 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5100220563
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dd414607d8f0a64f4fab3f3c59f0df699464f484 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741)
* 453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5100248916
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dd414607d8f0a64f4fab3f3c59f0df699464f484 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741)
* 453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5095037173
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f09d7f048606cb5f625bdec33492c7362effda Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735)
* dd414607d8f0a64f4fab3f3c59f0df699464f484 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5094988390
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f09d7f048606cb5f625bdec33492c7362effda Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735)
* dd414607d8f0a64f4fab3f3c59f0df699464f484 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] 19249 resolve required columns from table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-4970989657
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 50cf40647d30e797753d6b2815adfdfa4b3aedec 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5026966533
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 50cf40647d30e797753d6b2815adfdfa4b3aedec Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5030819638
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 74e1c977c82f3f88867bd5fdc9774d4366592e57 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5100481525
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5091928115
@wombatu-kun Can you please help to review this? I have a few more PRs to get the Trino integrations landed in a good shape.
I'd like to get this track of work done by this week.
Thank you!
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5102435422
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
}, {
"hash" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753)
* b98144b32be9bf69be0e012ce5b7a49b9c5eebae 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5096079327
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* dd414607d8f0a64f4fab3f3c59f0df699464f484 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662219964
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -192,33 +201,52 @@ public ConnectorPageSource createPageSource(
.withBloomFilter(useParquetBloomFilter(session))
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session))
.build();
- ConnectorPageSource dataPageSource = createPageSource(
- session,
- isBaseFileOnly ? dataColumnHandles : hudiMetaAndDataColumnHandles,
- hudiSplit,
- fileSystem.newInputFile(Location.of(hudiBaseFileOpt.get().getPath()), hudiBaseFileOpt.get().getFileSize()),
- hudiBaseFileOpt.get().getPath(),
- start,
- length,
- OptionalLong.of(hudiBaseFileOpt.get().getFileSize()),
- dataSourceStats,
- sessionOptions,
- timeZone, dynamicFilter, isBaseFileOnly);
-
- SynthesizedColumnHandler synthesizedColumnHandler = SynthesizedColumnHandler.create(hudiSplit);
+ PrefilledColumnValues prefilledColumnValues = PrefilledColumnValues.create(hudiSplit);
// Avoid avro serialization if split/filegroup only contains base files
if (isBaseFileOnly) {
return new HudiBaseFileOnlyPageSource(
- dataPageSource,
+ createBaseFilePageSource(session, dataColumnHandles, hudiSplit, fileSystem, sessionOptions, start, length, dynamicFilter, true),
hiveColumnHandles,
dataColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ // The merge path below is built around a base-file page source; fail log-only file slices with a
+ // clear error instead of an opaque NoSuchElementException from getBaseFile().orElseThrow().
+ // TODO: support log-only file slices by feeding the file-group reader an empty base page source.
+ if (hudiBaseFileOpt.isEmpty()) {
+ throw new TrinoException(NOT_SUPPORTED, "Hudi splits with log files but no base file are not supported: "
+ + hudiSplit.getLogFiles().getFirst().getPath());
}
// TODO: Move this into HudiTableHandle
HoodieTableMetaClient metaClient = buildTableMetaClient(
fileSystemFactory.create(session), hudiTableHandle.getSchemaTableName().toString(), hudiTableHandle.getBasePath());
+ HoodieSchema dataSchema =
+ Optional.ofNullable(hudiTableHandle.getTableSchema())
+ .orElseGet(() -> getLatestTableSchema(metaClient, hudiTableHandle.getTableName()));
+ TypedProperties readerProps = buildReaderProperties(session, metaClient);
+
+ // A non-projection-compatible CUSTOM merger makes the file-group reader demand the FULL table
+ // schema as requiredSchema for this split (it has log files), for the base and log reads alike.
+ // Expand the read projection to the full schema up front so the base page source carries every
+ // merge column; the log page sources resolve their columns from the same expanded handles.
+ List<HiveColumnHandle> readColumnHandles;
+ if (requiresFullSchemaRead(metaClient.getTableConfig(), readerProps)) {
+ log.debug("Expanding the read projection of %s to the full table schema: the resolved record merger is not projection compatible",
+ hudiTableHandle.getSchemaTableName());
+ readColumnHandles = appendMissingSchemaColumns(dataSchema, hudiMetaAndDataColumnHandles);
+ }
+ else {
+ // The metastore may lack merge-required columns the table schema carries (e.g. hive sync with
+ // omit_metadata_fields=true drops _hoodie_operation); recover them from the already-resolved
+ // schema so the base read is not starved of them.
+ readColumnHandles = appendMissingMergeRequiredColumns(dataSchema, hudiMetaAndDataColumnHandles, metaClient.getTableConfig());
Review Comment:
Nothing fails if this line goes back to `readColumnHandles = hudiMetaAndDataColumnHandles` - `TestHudiMergeRequiredColumns` only calls the helper directly, and every fixture's metastore column list matches its Avro schema exactly. Worth one fixture whose metastore omits a merge-required column its schema carries, so the recovery is pinned end to end.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +622,73 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} keeps only metastore data
+ * columns, and the merge read path recovers any name the metastore does not carry from the resolved
+ * table schema ({@link #appendMissingMergeRequiredColumns}) -- the gate the file-group reader itself
+ * applies -- so tables whose schema has neither field read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
+ requiredColumnNames.add(OPERATION_METADATA_FIELD);
+ String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
+ String deleteMarker = tableConfig.getProps().getProperty(DELETE_MARKER);
+ // DeleteContext only honors a custom delete key when the marker value is also set
+ if (!StringUtils.isNullOrEmpty(deleteKey) && !StringUtils.isNullOrEmpty(deleteMarker)) {
+ requiredColumnNames.add(deleteKey);
+ }
+ return requiredColumnNames;
+ }
+
+ /**
+ * Returns {@code projection} extended with handles for the merge-required column names it does not
+ * already carry (matched case-insensitively), typed from {@code dataSchema} via {@link #toColumnHandle};
+ * names absent from the schema as well are dropped. The Avro table schema is the gate the file-group
Review Comment:
The file-group reader schema-gates only `_hoodie_operation` and `_hoodie_is_deleted`; ordering fields, record-key fields and the custom delete key it adds unconditionally. Worth scoping the sentence to those two, since this method gates all five.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,291 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Column;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.metastore.PrincipalPrivileges;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.plugin.hudi.HudiConnector;
+import io.trino.spi.security.ConnectorIdentity;
+import io.trino.testing.QueryRunner;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_REALTIME_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS;
+import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Shared machinery for the non-partitioned Merge-On-Read fixtures that exist to exercise record merging at
+ * read time. The table is written by the Hudi Java write client into a local staging directory and then
+ * mirrored into the Trino filesystem the connector reads from; two metastore tables are registered against
+ * that location, a read-optimized one (base files only) and a real-time one (suffix {@code _rt}) that merges
+ * base + log files through the file group reader.
+ * <p>
+ * Inline compaction is off for every fixture so the log files written by the delta commits survive and must
+ * be merged at read time, and the metadata table is off because MDT writes need hbase dependencies that are
+ * not on the Trino test classpath.
+ * <p>
+ * Subclasses supply the schema, the merge-related table and write configuration, and the commits; everything
+ * a fixture does not vary lives here.
+ */
+public abstract class AbstractMergerHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ /** Every fixture built on this base is keyed on {@code key} and ordered by {@code ts}, in a single unnamed partition. */
+ protected static final String RECORD_KEY_FIELD = "key";
+ protected static final String ORDERING_FIELD = "ts";
+
+ private static final String PARTITION_PATH = "";
+
+ /** Hudi metadata columns, prepended to every table's data columns in the metastore definition. */
+ private static final List<Column> HUDI_META_COLUMNS = ImmutableList.of(
+ new Column("_hoodie_commit_time", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_commit_seqno", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_record_key", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_partition_path", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_file_name", HIVE_STRING, Optional.empty(), Map.of()));
+
+ private final String tableName;
+
+ private TrinoFileSystem fileSystem;
+ private Location tableLocation;
+ private java.nio.file.Path stagingDir;
+ private Path stagingTablePath;
+ private HoodieJavaWriteClient<HoodieAvroPayload> writeClient;
+
+ protected AbstractMergerHudiTablesInitializer(String tableName)
+ {
+ this.tableName = requireNonNull(tableName, "tableName is null");
+ }
+
+ @Override
+ public final void initializeTables(QueryRunner queryRunner, Location externalLocation, String schemaName)
+ throws Exception
+ {
+ fileSystem = ((HudiConnector) queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(TrinoFileSystemFactory.class)
+ .create(ConnectorIdentity.ofUser("test"));
+ HiveMetastore metastore = ((HudiConnector) queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(HiveMetastoreFactory.class)
+ .createMetastore(Optional.empty());
+
+ tableLocation = externalLocation.appendPath(tableName);
+ stagingDir = createTempDirectory(tableName.replace('_', '-'));
+ stagingTablePath = new Path(stagingDir.resolve(tableName).toUri());
+
+ boolean initialized = false;
+ try {
+ initTable();
+ afterTableInit();
+ writeClient = createWriteClient();
+ writeInitialCommits(writeClient);
+ syncToTrino();
+
+ metastore.createTable(createTableDefinition(schemaName, tableName, false), PrincipalPrivileges.NO_PRIVILEGES);
+ metastore.createTable(createTableDefinition(schemaName, tableName + "_rt", true), PrincipalPrivileges.NO_PRIVILEGES);
+ initialized = true;
+ }
+ finally {
+ // Only fixtures that keep writing commits after initialization need the staging directory and the
+ // write client to outlive this call; they are responsible for calling close().
+ if (!initialized || !keepsWriterOpen()) {
+ close();
+ }
+ }
+ }
+
+ /** The table's data columns, in schema order; the Hudi metadata columns are prepended by this class. */
+ protected abstract List<Column> dataColumns();
+
+ /** The Avro schema the write client writes, matching {@link #dataColumns()}. */
+ protected abstract Schema avroSchema();
+
+ /**
+ * Applies the fixture's merge-related table configuration (merge mode, merge strategy id, payload class).
+ * The table type, record key fields and ordering fields are set by this class.
+ */
+ protected abstract void configureTableConfig(HoodieTableMetaClient.TableBuilder tableBuilder);
+
+ /** Applies the fixture's merge-related write configuration (merge mode, merge strategy id, merger impl classes). */
+ protected abstract void configureWriteConfig(HoodieWriteConfig.Builder writeConfigBuilder);
+
+ /** Writes the commits that make up the fixture's initial state. */
+ protected abstract void writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException;
+
+ /** Runs on the freshly created table, before the write client exists and before any data is written. */
+ protected void afterTableInit()
+ throws IOException {}
+
+ /** Inline compaction is disabled, so this only has to stay above the number of delta commits a fixture writes. */
+ protected int maxDeltaCommitsBeforeCompaction()
+ {
+ return 100;
+ }
+
+ /** Whether the staging directory and write client survive {@link #initializeTables}; such fixtures must call {@link #close()}. */
+ protected boolean keepsWriterOpen()
+ {
+ return false;
+ }
+
+ public void close()
+ throws IOException
+ {
+ if (writeClient != null) {
+ writeClient.close();
+ writeClient = null;
+ }
+ if (stagingDir != null) {
+ deleteRecursively(stagingDir, ALLOW_INSECURE);
+ stagingDir = null;
+ }
+ }
+
+ /** Local directory the write client writes into, before {@link #syncToTrino()} mirrors it to the connector. */
+ protected java.nio.file.Path stagingTableDirectory()
+ {
+ return stagingDir.resolve(tableName);
+ }
+
+ protected HoodieJavaWriteClient<HoodieAvroPayload> writeClient()
+ {
+ return writeClient;
+ }
+
+ protected static HoodieRecord<HoodieAvroPayload> avroRecord(GenericRecord record, String key)
+ {
+ return new HoodieAvroRecord<>(new HoodieKey(key, PARTITION_PATH), new HoodieAvroPayload(Option.of(record)), null);
+ }
+
+ /** Mirrors the staged table into the Trino filesystem so the connector observes the commits written so far. */
+ protected void syncToTrino()
+ {
+ try {
+ if (fileSystem.directoryExists(tableLocation).orElse(false)) {
+ fileSystem.deleteDirectory(tableLocation);
+ }
+ ResourceHudiTablesInitializer.copyDir(stagingTableDirectory(), fileSystem, tableLocation);
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Failed to sync staged Hudi table to Trino filesystem", e);
+ }
+ }
+
+ private void initTable()
+ {
+ HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.MERGE_ON_READ)
+ .setTableName(tableName)
+ .setTimelineLayoutVersion(1)
+ .setBootstrapIndexClass(NoOpBootstrapIndex.class.getName())
+ .setRecordKeyFields(RECORD_KEY_FIELD)
+ .setOrderingFields(ORDERING_FIELD);
+ configureTableConfig(tableBuilder);
+ try {
+ tableBuilder.initTable(new HadoopStorageConfiguration(new Configuration()), stagingTablePath.toString());
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Could not init table " + tableName, e);
+ }
+ }
+
+ private HoodieJavaWriteClient<HoodieAvroPayload> createWriteClient()
+ {
+ Configuration conf = new Configuration();
+ HoodieWriteConfig.Builder writeConfigBuilder = HoodieWriteConfig.newBuilder()
+ .withPath(stagingTablePath.toString())
+ .withSchema(avroSchema().toString())
+ .withParallelism(2, 2)
+ .withDeleteParallelism(2)
+ .forTable(tableName)
+ .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ // Keep log files around so merging runs at read time.
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommitsBeforeCompaction())
+ .build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ // MDT writes require hbase deps not present in the Trino runtime.
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build());
+ // The ordering field is carried by the table config (setOrderingFields); the deprecated
Review Comment:
This comment used to sit inside the builder chain where `withPreCombineField` would have gone; above `configureWriteConfig` it reads as documentation of a hook that has nothing to do with pre-combine. Worth moving it back into the chain or dropping it.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/PayloadOnlyMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at TABLE VERSION 6 whose only merge configuration is a
+ * {@link org.apache.hudi.common.model.HoodieRecordPayload} class ({@link RankBasedTestPayload}), the way a
+ * genuine pre-1.0 table looks on storage.
+ * <p>
+ * Such a table resolves at read time to {@code CUSTOM} merge mode with the payload-based merge strategy, which
+ * in turn resolves {@code HoodieAvroRecordMerger}. That merger is not projection compatible, so the file group
+ * reader demands the FULL table schema for base and log reads alike and nothing prepends the payload's decision
+ * column into the connector's read projection.
+ * <p>
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of the same keys (log files), with data
+ * laid out so each merge direction is distinguishable: {@code k1}'s winning rank is on the LOG record and
+ * {@code k2}'s on the BASE record. See {@code TestHudiNonProjectionCompatibleMerger}.
+ */
+public class PayloadOnlyMergerHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "payload_only_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String RANK_FIELD = RankBasedTestPayload.RANK_COLUMN;
+ private static final HoodieTableVersion TABLE_VERSION = HoodieTableVersion.SIX;
+
+ public PayloadOnlyMergerHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("name", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("value", HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(RANK_FIELD, HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(), Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD, Schema.create(Schema.Type.STRING)),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(RANK_FIELD, Schema.create(Schema.Type.LONG)),
+ new Schema.Field(ORDERING_FIELD, Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder tableBuilder)
+ {
+ // Only the payload class: the merge mode and the merge strategy id must stay out of hoodie.properties
+ // so the reader has to infer them, which is what makes this a pre-1.0 payload-only table.
+ tableBuilder
+ .setTableVersion(TABLE_VERSION)
+ .setPayloadClassName(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder writeConfigBuilder)
+ {
+ writeConfigBuilder
+ .withWriteTableVersion(TABLE_VERSION.versionCode())
+ // Without this the write client silently upgrades the table to the current version on the first commit.
+ .withAutoUpgradeVersion(false)
+ .withWritePayLoad(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void afterTableInit()
+ throws IOException
+ {
+ stripInferredMergeConfigs();
+ }
+
+ @Override
+ protected void writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException
+ {
+ Schema schema = avroSchema();
+ // First commit: bulk insert base records (produces base parquet files).
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, 5L, 1L),
+ record(schema, "k2", "k2_base", 100L, 9L, 1L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Second commit: upserts the same keys (produces log files since inline compaction is disabled).
+ // k1 update has a HIGHER rank (7 > 5) -> the payload keeps the update (99): the LOG record's rank decides.
+ // k2 update has a LOWER rank (1 < 9) -> the payload keeps the base record (100): the BASE record's rank
+ // decides, which only works when the base read carries merge_rank despite it never being projected.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 99L, 7L, 2L),
+ record(schema, "k2", "k2_updated", 4L, 1L, 2L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // The commits go through the table config; re-check that the writer left the fixture payload-only.
+ stripInferredMergeConfigs();
+ }
+
+ /**
+ * Removes the merge mode and merge strategy id that table creation infers and persists for every pre-v9
+ * table, then verifies what is left. The fixture has to mimic a genuine pre-1.0 table, which persists its
+ * payload class and nothing else about merging; leaving the inferred keys in would hand the reader the
+ * answer this fixture exists to make it derive.
+ */
+ private void stripInferredMergeConfigs()
+ throws IOException
+ {
+ Path metaDirectory = stagingTableDirectory().resolve(".hoodie");
+ Path propertiesFile = metaDirectory.resolve("hoodie.properties");
+ List<String> retainedLines = Files.readAllLines(propertiesFile, UTF_8).stream()
+ .filter(line -> !isEntryFor(line, HoodieTableConfig.RECORD_MERGE_MODE.key())
Review Comment:
Table creation never persists `hoodie.record.merge.mode` for a version 6 table - `HoodieTableConfig.dropInvalidConfigs` drops it because the config's since-version is 1.0.0 - so only the strategy id is ever written and this filter plus its `checkAbsent` assert nothing. Worth dropping the merge-mode half, or rewording the javadoc, which says both are persisted.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CompositeHudiTablesInitializer.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.filesystem.Location;
+import io.trino.testing.QueryRunner;
+
+import java.util.List;
+
+/**
+ * Runs several {@link HudiTablesInitializer}s against one query runner, in order, so a single test class can
+ * load more than one table fixture. Each delegate must create its own tables under a distinct name.
+ */
+public class CompositeHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ private final List<HudiTablesInitializer> delegates;
+
+ public CompositeHudiTablesInitializer(HudiTablesInitializer... delegates)
+ {
+ this.delegates = ImmutableList.copyOf(delegates);
+ }
+
+ @Override
+ public void initializeTables(QueryRunner queryRunner, Location externalLocation, String schemaName)
+ throws Exception
+ {
+ for (HudiTablesInitializer delegate : delegates) {
+ delegate.initializeTables(queryRunner, externalLocation, schemaName);
Review Comment:
A delegate that returns `keepsWriterOpen()` true keeps its write client and staging directory alive, and the composite has no way to release them since `HudiTablesInitializer` declares no close. Worth saying so in the javadoc, or making the interface `AutoCloseable` with a no-op default so the composite can forward it.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,135 @@ public static Schema.Field getFieldFromSchema(String columnName, Schema schema)
"Failed to get column " + columnName + " from table schema");
}
+ /**
+ * Builds a {@link HiveColumnHandle} for a table-schema field, typing it from the field's Avro schema.
+ * Used to resolve columns the file-group reader requires for merging but the connector projection does
+ * not carry (e.g. {@code _hoodie_commit_time} or a delete-marker column on a narrow query).
+ * <p>
+ * The handle's {@code hiveColumnIndex} is a placeholder: page sources built from these handles resolve
+ * parquet columns by NAME (directly when {@code hudi.parquet.use-column-names=true}, otherwise
+ * {@link HudiPageSourceProvider#remapColumnIndicesToPhysical} rebuilds every index from the file
+ * schema by name), so the ordinal is never used to locate data.
+ * <p>
+ * Avro types whose Trino mapping has no Hive counterpart (uuid, time-millis/micros) fail with
+ * NOT_SUPPORTED from {@link HiveTypeTranslator#toHiveType}; such columns are equally unreadable
+ * through the metastore schema, so this surfaces the same limitation with a clear error.
+ */
+ public static HiveColumnHandle toColumnHandle(HoodieSchemaField field)
+ {
+ Type trinoType;
+ try {
+ trinoType = AVRO_TYPE_HANDLER.typeFor(field.schema().toAvroSchema());
+ }
+ catch (AvroTypeException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to map Avro type of column " + field.name() + " to a Trino type", e);
+ }
+ return new HiveColumnHandle(
+ field.name(),
+ 0,
+ HiveTypeTranslator.toHiveType(trinoType),
+ trinoType,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /**
+ * Resolves the merge mode and merge strategy id the file-group reader will use for this table,
+ * applying hudi-common's version-gated inference: the merge MODE is inferred for any table below
+ * version 9 ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}), while the STRATEGY ID
+ * the merger is resolved with is inferred only below version 8
+ * ({@code HoodieReaderContext.initRecordMerger}). The asymmetry is deliberate and must mirror
+ * hudi-common exactly: pre-v9 tables may persist neither config (a 0.x table with a custom payload
+ * class infers CUSTOM mode with the payload-based strategy id), so a connector-side decision built
+ * on the raw configs would disagree with the file-group reader's.
+ */
+ public static Pair<RecordMergeMode, String> resolveMergeModeAndStrategyId(HoodieTableConfig tableConfig)
+ {
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode();
+ String mergeStrategyId = tableConfig.getRecordMergeStrategyId();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)) {
+ Triple<RecordMergeMode, String, String> inferred = HoodieTableConfig.inferMergingConfigsForPreV9Table(
+ tableConfig.getRecordMergeMode(),
+ tableConfig.getPayloadClass(),
+ tableConfig.getRecordMergeStrategyId(),
+ tableConfig.getOrderingFieldsStr().orElse(null),
+ tableConfig.getTableVersion());
+ mergeMode = inferred.getLeft();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ mergeStrategyId = inferred.getRight();
+ }
+ }
+ return Pair.of(mergeMode, mergeStrategyId);
+ }
+
+ /**
+ * Returns whether reads of this table go through a CUSTOM record merger that is NOT projection
+ * compatible. For such mergers the file-group reader demands the FULL table schema as its required
+ * schema on any split with log files ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}),
+ * so the connector must read every table column, not just the query projection.
+ * <p>
+ * Callers must pass the merge mode and strategy id resolved by {@link #resolveMergeModeAndStrategyId}
+ * so the {@link HoodieRecordUtils#createValidRecordMerger} call here sees the same inputs as the
+ * file-group reader's own resolution. A CUSTOM mode without a strategy id returns false: no projection
+ * expansion is possible without a merger to ask, and {@code HudiTrinoReaderContext.getRecordMerger}
+ * rejects the read with an actionable error when the merger is actually needed -- hudi-common's
+ * {@link HoodieRecordUtils#createValidRecordMerger} would otherwise NPE on the null id.
+ * <p>
+ * Note the payload-based strategy id resolves {@code HoodieAvroRecordMerger}, which keeps the
+ * interface default {@code isProjectionCompatible() == false} -- so EVERY custom-payload table takes
+ * the full-schema read path, not just tables with exotic custom mergers. That matches the file-group
+ * reader (and Spark); the extra I/O is inherent to payload-based merging.
+ */
+ public static boolean usesNonProjectionCompatibleMerger(RecordMergeMode mergeMode, String mergeStrategyId, String mergeImplClasses)
+ {
+ if (mergeMode != RecordMergeMode.CUSTOM || StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ return false;
+ }
+ Option<HoodieRecordMerger> merger = HoodieRecordUtils.createValidRecordMerger(EngineType.JAVA, mergeImplClasses, mergeStrategyId);
+ return merger.isPresent() && !merger.get().isProjectionCompatible();
+ }
+
+ /**
+ * Rejects a CUSTOM-mode read that has no merge strategy id to resolve a record merger with. The
+ * strategy id is only inferred below table version 8 ({@link #resolveMergeModeAndStrategyId}), so a
+ * version 8 table written without a persisted id reaches merger resolution with a null id, which
+ * {@link HoodieRecordUtils#createValidRecordMerger} dereferences straight away.
+ */
+ public static void validateCustomMergeStrategyId(String mergeStrategyId)
+ {
+ if (StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ String strategyIdKey = HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key();
+ throw new TrinoException(HUDI_BAD_DATA,
+ ("Table resolved to %s merge mode but persists no `%s`, so no record merger can be resolved for merging log files. "
+ + "Table version 8 tables written without a persisted strategy id resolve this way; set `%s` on the table to the "
Review Comment:
The strategy id is passed through unchanged from table version 8 upward, so a version 9 or 10 table whose config says CUSTOM without an id lands here too and gets told this is a version 8 situation. Worth widening the message and the javadoc to "version 8 or later".
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.trino.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiUtil.appendMissingMergeRequiredColumns;
+import static io.trino.plugin.hudi.HudiUtil.mergeRequiredColumnNames;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY;
+import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER;
+import static org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.OPERATION_METADATA_FIELD;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests {@link HudiUtil#mergeRequiredColumnNames}, which mirrors the non-CUSTOM branch of the file-group
+ * reader's {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging} so the base-file read
+ * projection carries every column the merge may consult (ordering fields, delete markers, the operation
+ * field, record-key data columns) even when the query does not project them, plus the merge-path recovery
+ * of names the metastore could not resolve ({@link HudiUtil#appendMissingMergeRequiredColumns}).
+ */
+class TestHudiMergeRequiredColumns
+{
+ @Test
+ public void testOrderingFieldsFollowMergeMode()
+ {
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.ORDERING_FIELDS, "ts,seq");
+
+ assertThat(mergeRequiredColumnNames(tableConfig, RecordMergeMode.EVENT_TIME_ORDERING))
+ .contains("ts", "seq");
+ // Commit-time merging ignores ordering fields
+ assertThat(mergeRequiredColumnNames(tableConfig, RecordMergeMode.COMMIT_TIME_ORDERING))
+ .doesNotContain("ts", "seq");
+ assertThat(mergeRequiredColumnNames(tableConfig, null))
+ .doesNotContain("ts", "seq");
+ }
+
+ @Test
+ public void testDeleteAndOperationColumnsAlwaysRequested()
+ {
+ // Requested unconditionally: getMergeRequiredColumnHandles drops names that are in neither the
Review Comment:
`getMergeRequiredColumnHandles` drops every name the metastore lacks, including names the Avro table schema does carry - the schema half of the gate lives in `appendMissingMergeRequiredColumns`, on the page-source path. Worth splitting the claim across the two, since as written it describes a union gate that no single function implements.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662819839
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -192,33 +201,52 @@ public ConnectorPageSource createPageSource(
.withBloomFilter(useParquetBloomFilter(session))
.withVectorizedDecodingEnabled(isParquetVectorizedDecodingEnabled(session))
.build();
- ConnectorPageSource dataPageSource = createPageSource(
- session,
- isBaseFileOnly ? dataColumnHandles : hudiMetaAndDataColumnHandles,
- hudiSplit,
- fileSystem.newInputFile(Location.of(hudiBaseFileOpt.get().getPath()), hudiBaseFileOpt.get().getFileSize()),
- hudiBaseFileOpt.get().getPath(),
- start,
- length,
- OptionalLong.of(hudiBaseFileOpt.get().getFileSize()),
- dataSourceStats,
- sessionOptions,
- timeZone, dynamicFilter, isBaseFileOnly);
-
- SynthesizedColumnHandler synthesizedColumnHandler = SynthesizedColumnHandler.create(hudiSplit);
+ PrefilledColumnValues prefilledColumnValues = PrefilledColumnValues.create(hudiSplit);
// Avoid avro serialization if split/filegroup only contains base files
if (isBaseFileOnly) {
return new HudiBaseFileOnlyPageSource(
- dataPageSource,
+ createBaseFilePageSource(session, dataColumnHandles, hudiSplit, fileSystem, sessionOptions, start, length, dynamicFilter, true),
hiveColumnHandles,
dataColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ // The merge path below is built around a base-file page source; fail log-only file slices with a
+ // clear error instead of an opaque NoSuchElementException from getBaseFile().orElseThrow().
+ // TODO: support log-only file slices by feeding the file-group reader an empty base page source.
+ if (hudiBaseFileOpt.isEmpty()) {
+ throw new TrinoException(NOT_SUPPORTED, "Hudi splits with log files but no base file are not supported: "
+ + hudiSplit.getLogFiles().getFirst().getPath());
}
// TODO: Move this into HudiTableHandle
HoodieTableMetaClient metaClient = buildTableMetaClient(
fileSystemFactory.create(session), hudiTableHandle.getSchemaTableName().toString(), hudiTableHandle.getBasePath());
+ HoodieSchema dataSchema =
+ Optional.ofNullable(hudiTableHandle.getTableSchema())
+ .orElseGet(() -> getLatestTableSchema(metaClient, hudiTableHandle.getTableName()));
+ TypedProperties readerProps = buildReaderProperties(session, metaClient);
+
+ // A non-projection-compatible CUSTOM merger makes the file-group reader demand the FULL table
+ // schema as requiredSchema for this split (it has log files), for the base and log reads alike.
+ // Expand the read projection to the full schema up front so the base page source carries every
+ // merge column; the log page sources resolve their columns from the same expanded handles.
+ List<HiveColumnHandle> readColumnHandles;
+ if (requiresFullSchemaRead(metaClient.getTableConfig(), readerProps)) {
+ log.debug("Expanding the read projection of %s to the full table schema: the resolved record merger is not projection compatible",
+ hudiTableHandle.getSchemaTableName());
+ readColumnHandles = appendMissingSchemaColumns(dataSchema, hudiMetaAndDataColumnHandles);
+ }
+ else {
+ // The metastore may lack merge-required columns the table schema carries (e.g. hive sync with
+ // omit_metadata_fields=true drops _hoodie_operation); recover them from the already-resolved
+ // schema so the base read is not starved of them.
+ readColumnHandles = appendMissingMergeRequiredColumns(dataSchema, hudiMetaAndDataColumnHandles, metaClient.getTableConfig());
Review Comment:
Added exactly that: `omitted_ordering_field_mor`, an EVENT_TIME_ORDERING MOR fixture whose metastore column list omits the `ts` its Avro schema carries. Row-level and narrow-projection `sum(value) = 199` cases in `TestHudiNonProjectionCompatibleMerger`, and I verified the pin the way you framed it: with the line reverted to `hudiMetaAndDataColumnHandles`, both fail with the missing-column guard error.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5091996766
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6569ea38b806659ffd0e192334fa2f4dd4609efc Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733)
* 65f09d7f048606cb5f625bdec33492c7362effda Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5102486520
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
}, {
"hash" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15765",
"triggerID" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753)
* b98144b32be9bf69be0e012ce5b7a49b9c5eebae Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15765)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5027153097
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "PENDING",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 50cf40647d30e797753d6b2815adfdfa4b3aedec Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483)
* 74e1c977c82f3f88867bd5fdc9774d4366592e57 Azure: [PENDING](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3658583976
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,115 @@ public static Schema.Field getFieldFromSchema(String columnName, Schema schema)
"Failed to get column " + columnName + " from table schema");
}
+ /**
+ * Builds a {@link HiveColumnHandle} for a table-schema field, typing it from the field's Avro schema.
+ * Used to resolve columns the file-group reader requires for merging but the connector projection does
+ * not carry (e.g. {@code _hoodie_commit_time} or a delete-marker column on a narrow query).
+ * <p>
+ * The handle's {@code hiveColumnIndex} is a placeholder: page sources built from these handles resolve
+ * parquet columns by NAME (directly when {@code hudi.parquet.use-column-names=true}, otherwise
+ * {@link HudiPageSourceProvider#remapColumnIndicesToPhysical} rebuilds every index from the file
+ * schema by name), so the ordinal is never used to locate data.
+ * <p>
+ * Avro types whose Trino mapping has no Hive counterpart (uuid, time-millis/micros) fail with
+ * NOT_SUPPORTED from {@link HiveTypeTranslator#toHiveType}; such columns are equally unreadable
+ * through the metastore schema, so this surfaces the same limitation with a clear error.
+ */
+ public static HiveColumnHandle toColumnHandle(HoodieSchemaField field)
+ {
+ Type trinoType;
+ try {
+ trinoType = AVRO_TYPE_HANDLER.typeFor(field.schema().toAvroSchema());
+ }
+ catch (AvroTypeException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to map Avro type of column " + field.name() + " to a Trino type", e);
+ }
+ return new HiveColumnHandle(
+ field.name(),
+ 0,
+ HiveTypeTranslator.toHiveType(trinoType),
+ trinoType,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /**
+ * Resolves the merge mode and merge strategy id the file-group reader will use for this table,
+ * applying hudi-common's version-gated inference: the merge MODE is inferred for any table below
+ * version 9 ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}), while the STRATEGY ID
+ * the merger is resolved with is inferred only below version 8
+ * ({@code HoodieReaderContext.initRecordMerger}). The asymmetry is deliberate and must mirror
+ * hudi-common exactly: pre-v9 tables may persist neither config (a 0.x table with a custom payload
+ * class infers CUSTOM mode with the payload-based strategy id), so a connector-side decision built
+ * on the raw configs would disagree with the file-group reader's.
+ */
+ public static Pair<RecordMergeMode, String> resolveMergeModeAndStrategyId(HoodieTableConfig tableConfig)
+ {
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode();
+ String mergeStrategyId = tableConfig.getRecordMergeStrategyId();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)) {
+ Triple<RecordMergeMode, String, String> inferred = HoodieTableConfig.inferMergingConfigsForPreV9Table(
+ tableConfig.getRecordMergeMode(),
+ tableConfig.getPayloadClass(),
+ tableConfig.getRecordMergeStrategyId(),
+ tableConfig.getOrderingFieldsStr().orElse(null),
+ tableConfig.getTableVersion());
+ mergeMode = inferred.getLeft();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ mergeStrategyId = inferred.getRight();
+ }
+ }
+ return Pair.of(mergeMode, mergeStrategyId);
+ }
+
+ /**
+ * Returns whether reads of this table go through a CUSTOM record merger that is NOT projection
+ * compatible. For such mergers the file-group reader demands the FULL table schema as its required
+ * schema on any split with log files ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}),
+ * so the connector must read every table column, not just the query projection.
+ * <p>
+ * Callers must pass the merge mode and strategy id resolved by {@link #resolveMergeModeAndStrategyId}
+ * so the {@link HoodieRecordUtils#createValidRecordMerger} call here sees the same inputs as the
+ * file-group reader's own resolution. A CUSTOM mode without a strategy id returns false: no merger
+ * can be resolved from it, and the file-group reader itself then fails loudly.
Review Comment:
For a null strategy id the file-group reader does not fail loudly - `createValidRecordMerger` dereferences the id on its first line and NPEs, which is the v8-plus-custom-payload case the new unit test pins. Either drop the "fails loudly" half or raise the actionable error here.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -425,10 +488,14 @@ public static List<HiveColumnHandle> remapColumnIndicesToPhysical(
String requestedName = originalHandle.getBaseColumnName();
// Determine the key to use for looking up the physical index
- String lookupKey = caseSensitive ? requestedName : requestedName.toLowerCase(Locale.getDefault());
+ String lookupKey = caseSensitive ? requestedName : requestedName.toLowerCase(Locale.ROOT);
- // Find the physical index from the file schema map constructed from fielSchema
+ // Find the physical index from the file schema map constructed from fileSchema
Integer physicalIndex = physicalIndexMap.get(lookupKey);
+ if (physicalIndex == null) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR, format(
+ "Column '%s' not found in parquet file schema %s", requestedName, fileSchema.getName()));
Review Comment:
With the projection now widened to the full table schema, this throw makes `hudi.parquet.use-column-names=false` fail on any base file written before a column was added, while the default name-based path null-fills it in `createParquetPageSource`. Mapping a not-found column to `fileFields.size()` would fall into the parquet reader's own absent-column handling instead - or is the hard failure deliberate here?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -260,17 +281,59 @@ public ConnectorPageSource createPageSource(
.withDataSchema(dataSchema)
.withRequestedSchema(HoodieSchema.fromAvroSchema(requestedSchema))
.withLatestCommitTime(hudiTableHandle.getLatestCommitTime())
- .withProps(buildReaderProperties(session, metaClient))
+ .withProps(readerProps)
.withShouldUseRecordPosition(false)
.withStart(start)
.withLength(length)
.build();
return new HudiPageSource(
dataPageSource,
fileGroupReader,
- readerContext,
hiveColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ private ConnectorPageSource createBaseFilePageSource(
+ ConnectorSession session,
+ List<HiveColumnHandle> columns,
+ HudiSplit hudiSplit,
+ TrinoFileSystem fileSystem,
+ ParquetReaderOptions sessionOptions,
+ long start,
+ long length,
+ DynamicFilter dynamicFilter,
+ boolean enablePredicatePushDown)
+ {
+ HudiBaseFile baseFile = hudiSplit.getBaseFile().orElseThrow();
+ return createPageSource(
+ session,
+ columns,
+ hudiSplit,
+ fileSystem.newInputFile(Location.of(baseFile.getPath()), baseFile.getFileSize()),
+ baseFile.getPath(),
+ start,
+ length,
+ OptionalLong.of(baseFile.getFileSize()),
+ dataSourceStats,
+ sessionOptions,
+ timeZone,
+ dynamicFilter,
+ enablePredicatePushDown);
+ }
+
+ /**
+ * Mirrors {@code FileGroupReaderSchemaHandler.generateRequiredSchema}'s decision for CUSTOM merge
+ * mode: resolves the merge mode and strategy id with the version-gated inference the file-group
+ * reader applies ({@link HudiUtil#resolveMergeModeAndStrategyId}) and asks the same resolved merger
+ * whether it is projection compatible, so the connector's read projection and the file-group
+ * reader's required schema cannot disagree.
Review Comment:
This says the two "cannot disagree", but only the CUSTOM and `isProjectionCompatible` half is mirrored exactly - the mandatory-fields half is a superset, and the guard this PR adds to `getFileRecordIterator` exists for exactly that drift. Worth softening to what the method actually guarantees.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiNonProjectionCompatibleMerger.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.NonProjectionCompatibleRankMerger;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.RT_TABLE_NAME;
+import static io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.TABLE_NAME;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Acceptance test for full-table-schema merge reads (apache/hudi#19249, issue comment on scope):
+ * {@link NonProjectionCompatibleRankMerger} does NOT override {@code isProjectionCompatible()} (default
+ * {@code false}) and does NOT declare {@code merge_rank} mandatory, so the file-group reader demands the
+ * FULL table schema as its required schema for base and log reads alike, and nothing prepends
+ * {@code merge_rank} into the connector's read projection.
+ * <p>
+ * The queries below never project {@code merge_rank}. The data is laid out so each merge direction is
+ * proven independently: {@code k1}'s winning rank is on the LOG record (update wins, value 99) and
+ * {@code k2}'s winning rank is on the BASE record (base wins, value 100). A correct result therefore
+ * requires the un-projected rank column to be read on BOTH sides of the merge. {@code sum(value)} is a
+ * three-way discriminator: merged = 199, base-only = 110, built-in newest-wins = 103.
+ */
+public class TestHudiNonProjectionCompatibleMerger
Review Comment:
Every pre-v9 payload-class table now takes the full-table-schema read path, but nothing exercises it end to end - the v6 and v8 MOR fixtures all infer COMMIT_TIME or EVENT_TIME, and both CUSTOM initializers set an explicit strategy id and merger impl. Worth one table configured with only a payload class, so the payload-based strategy actually gets read through.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/NonProjectionCompatibleMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,242 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Column;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.metastore.PrincipalPrivileges;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.plugin.hudi.HudiConnector;
+import io.trino.spi.security.ConnectorIdentity;
+import io.trino.testing.QueryRunner;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_REALTIME_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS;
+import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS;
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at test runtime configured with the
+ * NON-projection-compatible {@link NonProjectionCompatibleRankMerger} via {@link RecordMergeMode#CUSTOM}.
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of the same keys (log files), with
+ * inline compaction disabled so the merger runs at read time over a full-table-schema read.
+ * <p>
+ * Data is laid out so each merge direction is distinguishable: one key's winning rank comes from the LOG
+ * record and the other's from the BASE record, so a correct result proves both sides of the merge supplied
+ * the (never projected) {@code merge_rank} column. See {@code TestHudiNonProjectionCompatibleMerger}.
+ */
+public class NonProjectionCompatibleMergerHudiTablesInitializer
Review Comment:
This is the third near-identical MoR initializer - `initializeTables`, `createTableDefinition`, the `_hoodie_*` column list and `createWriteClient` are verbatim copies of `CustomMergerHudiTablesInitializer` apart from names and the merger. Worth a shared base taking (tableName, mergerClass, strategyId, fields, records); follow-up, not a blocker.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +602,38 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} drops any name that is not a
+ * data column of the table, so tables without them read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
Review Comment:
These two are filtered against the metastore data columns by `buildColumnHandles`, while `getMandatoryFieldsForMerging` and `DeleteContext` gate them on the Avro table schema - a table whose metastore omits a field the schema has (hive-sync with `omit_metadata_fields=true` drops `_hoodie_operation`, a hand-written external DDL drops `_hoodie_is_deleted`) loses it from the base projection and the new guard then fails the read. Could these resolve off the table schema via `toColumnHandle`, the way the CUSTOM branch already does?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiFile;
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.RunLengthEncodedBlock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import static io.trino.metastore.Partitions.makePartName;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileModifiedTimeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileSizeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPartitionColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPathColumnHandle;
+import static io.trino.plugin.hive.util.HiveUtil.getPrefilledColumnValue;
+import static io.trino.spi.type.TypeUtils.writeNativeValue;
+
+/**
+ * Per-split constant values for the output columns that are not stored in the data file: Hive-style
+ * partition columns and Trino's hidden metadata columns ({@code $path}, {@code $file_size},
+ * {@code $file_modified_time}, {@code $partition}). Value computation delegates to
+ * {@link io.trino.plugin.hive.util.HiveUtil#getPrefilledColumnValue}, the same implementation the Hive
+ * connector uses for these columns (including the {@code "\N"} hive-null partition-value convention).
+ */
+public class PrefilledColumnValues
+{
+ private final Map<String, HivePartitionKey> partitionKeysByName;
+ private final String partitionName;
+ private final String filePath;
+ private final long fileSize;
+ private final long fileModifiedTime;
+
+ public static PrefilledColumnValues create(HudiSplit hudiSplit)
+ {
+ return new PrefilledColumnValues(hudiSplit);
+ }
+
+ private PrefilledColumnValues(HudiSplit hudiSplit)
+ {
+ List<HivePartitionKey> partitionKeys = hudiSplit.getPartitionKeys();
+ // ImmutableMap preserves insertion order, so $partition renders the keys in the split's
+ // partition-column order.
+ ImmutableMap.Builder<String, HivePartitionKey> byName = ImmutableMap.builder();
+ partitionKeys.forEach(partitionKey -> byName.put(partitionKey.name(), partitionKey));
+ this.partitionKeysByName = byName.buildOrThrow();
+ this.partitionName = makePartName(
+ partitionKeys.stream().map(HivePartitionKey::name).toList(),
+ partitionKeys.stream().map(HivePartitionKey::value).toList());
+ // Parquet files will be prioritised over log files
+ HudiFile hudiFile = hudiSplit.getBaseFile().isPresent()
+ ? hudiSplit.getBaseFile().get()
+ : hudiSplit.getLogFiles().getFirst();
+ this.filePath = hudiFile.getPath();
+ this.fileSize = hudiFile.getFileSize();
+ this.fileModifiedTime = hudiFile.getModificationTime();
+ }
+
+ /**
+ * Returns whether this split can provide a value for the column, i.e. it is a partition column of
+ * the split or a hidden metadata column Hudi populates.
+ */
+ public boolean isPrefilled(HiveColumnHandle columnHandle)
+ {
+ return partitionKeysByName.containsKey(columnHandle.getName())
+ || isPathColumnHandle(columnHandle)
+ || isFileSizeColumnHandle(columnHandle)
+ || isFileModifiedTimeColumnHandle(columnHandle)
+ || isPartitionColumnHandle(columnHandle);
+ }
+
+ /**
+ * Appends the column's value for this split to the builder; appends null for a column this split
+ * cannot provide.
+ */
+ public void appendTo(HiveColumnHandle columnHandle, BlockBuilder blockBuilder)
+ {
+ writeNativeValue(columnHandle.getType(), blockBuilder, nativeValueOf(columnHandle));
+ }
+
+ /**
+ * Builds a run-length-encoded {@link Block} repeating the column's constant value for
+ * {@code positionCount} positions (null-filled for a column this split cannot provide).
+ */
+ public Block toRleBlock(HiveColumnHandle columnHandle, int positionCount)
+ {
+ return RunLengthEncodedBlock.create(columnHandle.getType(), nativeValueOf(columnHandle), positionCount);
+ }
+
+ private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ if (!isPrefilled(columnHandle)) {
+ // Lenient null fill, e.g. for a hidden column Trino defines but Hudi does not populate.
+ return null;
+ }
+ HivePartitionKey partitionKey = partitionKeysByName.get(columnHandle.getName());
+ return getPrefilledColumnValue(
Review Comment:
The swap also moves `$file_modified_time` from a UTC-pinned zone to the worker's default zone - `getPrefilledColumnValue` packs with `DateTimeZone.getDefault()` where the deleted code used `UTC_KEY`. Same instant, but the rendered zone and zone-sensitive functions change, so it belongs in the delta list with the other three - or was UTC the accident?
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3659653729
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -260,17 +281,59 @@ public ConnectorPageSource createPageSource(
.withDataSchema(dataSchema)
.withRequestedSchema(HoodieSchema.fromAvroSchema(requestedSchema))
.withLatestCommitTime(hudiTableHandle.getLatestCommitTime())
- .withProps(buildReaderProperties(session, metaClient))
+ .withProps(readerProps)
.withShouldUseRecordPosition(false)
.withStart(start)
.withLength(length)
.build();
return new HudiPageSource(
dataPageSource,
fileGroupReader,
- readerContext,
hiveColumnHandles,
- synthesizedColumnHandler);
+ prefilledColumnValues);
+ }
+
+ private ConnectorPageSource createBaseFilePageSource(
+ ConnectorSession session,
+ List<HiveColumnHandle> columns,
+ HudiSplit hudiSplit,
+ TrinoFileSystem fileSystem,
+ ParquetReaderOptions sessionOptions,
+ long start,
+ long length,
+ DynamicFilter dynamicFilter,
+ boolean enablePredicatePushDown)
+ {
+ HudiBaseFile baseFile = hudiSplit.getBaseFile().orElseThrow();
+ return createPageSource(
+ session,
+ columns,
+ hudiSplit,
+ fileSystem.newInputFile(Location.of(baseFile.getPath()), baseFile.getFileSize()),
+ baseFile.getPath(),
+ start,
+ length,
+ OptionalLong.of(baseFile.getFileSize()),
+ dataSourceStats,
+ sessionOptions,
+ timeZone,
+ dynamicFilter,
+ enablePredicatePushDown);
+ }
+
+ /**
+ * Mirrors {@code FileGroupReaderSchemaHandler.generateRequiredSchema}'s decision for CUSTOM merge
+ * mode: resolves the merge mode and strategy id with the version-gated inference the file-group
+ * reader applies ({@link HudiUtil#resolveMergeModeAndStrategyId}) and asks the same resolved merger
+ * whether it is projection compatible, so the connector's read projection and the file-group
+ * reader's required schema cannot disagree.
Review Comment:
Reworded to what it actually guarantees: the CUSTOM/`isProjectionCompatible` decision is mirrored exactly, and the mandatory-fields side is now called out as a superset prediction with the `getFileRecordIterator` guard behind it.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiPageSourceProvider.java:
##########
@@ -425,10 +488,14 @@ public static List<HiveColumnHandle> remapColumnIndicesToPhysical(
String requestedName = originalHandle.getBaseColumnName();
// Determine the key to use for looking up the physical index
- String lookupKey = caseSensitive ? requestedName : requestedName.toLowerCase(Locale.getDefault());
+ String lookupKey = caseSensitive ? requestedName : requestedName.toLowerCase(Locale.ROOT);
- // Find the physical index from the file schema map constructed from fielSchema
+ // Find the physical index from the file schema map constructed from fileSchema
Integer physicalIndex = physicalIndexMap.get(lookupKey);
+ if (physicalIndex == null) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR, format(
+ "Column '%s' not found in parquet file schema %s", requestedName, fileSchema.getName()));
Review Comment:
The hard failure wasn't deliberate -- before this PR it was an autoboxing NPE, I had just given it a message. Went with your suggestion: a miss now maps to `fileFields.size()`. Checked the trino-hive 481 sources to be sure: all three consumers (`getParquetMessageType`, `createParquetPageSource`, `getParquetTupleDomain`) go through `getBaseColumnParquetType`, which treats an out-of-range index as absent, so the reader null-fills it exactly like the name-based path. Remap tests flipped to pin the mapping.
Null is also the right value here: schema-on-write never rewrites old base files, so a file predating a column add legitimately lacks it, and that is what the merge should see for those records. What neither path handles is full schema evolution (`hoodie.schema.on.read.enable`) -- a renamed column null-fills on both paths since nothing resolves through the internal schema. Filed #19381 to track that.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,115 @@ public static Schema.Field getFieldFromSchema(String columnName, Schema schema)
"Failed to get column " + columnName + " from table schema");
}
+ /**
+ * Builds a {@link HiveColumnHandle} for a table-schema field, typing it from the field's Avro schema.
+ * Used to resolve columns the file-group reader requires for merging but the connector projection does
+ * not carry (e.g. {@code _hoodie_commit_time} or a delete-marker column on a narrow query).
+ * <p>
+ * The handle's {@code hiveColumnIndex} is a placeholder: page sources built from these handles resolve
+ * parquet columns by NAME (directly when {@code hudi.parquet.use-column-names=true}, otherwise
+ * {@link HudiPageSourceProvider#remapColumnIndicesToPhysical} rebuilds every index from the file
+ * schema by name), so the ordinal is never used to locate data.
+ * <p>
+ * Avro types whose Trino mapping has no Hive counterpart (uuid, time-millis/micros) fail with
+ * NOT_SUPPORTED from {@link HiveTypeTranslator#toHiveType}; such columns are equally unreadable
+ * through the metastore schema, so this surfaces the same limitation with a clear error.
+ */
+ public static HiveColumnHandle toColumnHandle(HoodieSchemaField field)
+ {
+ Type trinoType;
+ try {
+ trinoType = AVRO_TYPE_HANDLER.typeFor(field.schema().toAvroSchema());
+ }
+ catch (AvroTypeException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to map Avro type of column " + field.name() + " to a Trino type", e);
+ }
+ return new HiveColumnHandle(
+ field.name(),
+ 0,
+ HiveTypeTranslator.toHiveType(trinoType),
+ trinoType,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /**
+ * Resolves the merge mode and merge strategy id the file-group reader will use for this table,
+ * applying hudi-common's version-gated inference: the merge MODE is inferred for any table below
+ * version 9 ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}), while the STRATEGY ID
+ * the merger is resolved with is inferred only below version 8
+ * ({@code HoodieReaderContext.initRecordMerger}). The asymmetry is deliberate and must mirror
+ * hudi-common exactly: pre-v9 tables may persist neither config (a 0.x table with a custom payload
+ * class infers CUSTOM mode with the payload-based strategy id), so a connector-side decision built
+ * on the raw configs would disagree with the file-group reader's.
+ */
+ public static Pair<RecordMergeMode, String> resolveMergeModeAndStrategyId(HoodieTableConfig tableConfig)
+ {
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode();
+ String mergeStrategyId = tableConfig.getRecordMergeStrategyId();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)) {
+ Triple<RecordMergeMode, String, String> inferred = HoodieTableConfig.inferMergingConfigsForPreV9Table(
+ tableConfig.getRecordMergeMode(),
+ tableConfig.getPayloadClass(),
+ tableConfig.getRecordMergeStrategyId(),
+ tableConfig.getOrderingFieldsStr().orElse(null),
+ tableConfig.getTableVersion());
+ mergeMode = inferred.getLeft();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ mergeStrategyId = inferred.getRight();
+ }
+ }
+ return Pair.of(mergeMode, mergeStrategyId);
+ }
+
+ /**
+ * Returns whether reads of this table go through a CUSTOM record merger that is NOT projection
+ * compatible. For such mergers the file-group reader demands the FULL table schema as its required
+ * schema on any split with log files ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}),
+ * so the connector must read every table column, not just the query projection.
+ * <p>
+ * Callers must pass the merge mode and strategy id resolved by {@link #resolveMergeModeAndStrategyId}
+ * so the {@link HoodieRecordUtils#createValidRecordMerger} call here sees the same inputs as the
+ * file-group reader's own resolution. A CUSTOM mode without a strategy id returns false: no merger
+ * can be resolved from it, and the file-group reader itself then fails loudly.
Review Comment:
Raised the error instead of softening the doc: the CUSTOM branch of `getRecordMerger` now rejects a null/empty strategy id with `HUDI_BAD_DATA` naming `hoodie.record.merge.strategy.id` before it reaches `createValidRecordMerger`. Javadoc updated to match.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +602,38 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} drops any name that is not a
+ * data column of the table, so tables without them read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
Review Comment:
Fixed, though not inside `getMergeRequiredColumnHandles` itself -- resolving the table schema there costs a timeline read on every query, and the `*CacheFileOperations` tests caught that immediately. Instead the merge read path now recovers any merge-required name the metastore couldn't resolve from the already-resolved `dataSchema` via `toColumnHandle` (`appendMissingMergeRequiredColumns`), so the gate matches `getMandatoryFieldsForMerging`/`DeleteContext` and no query pays extra I/O. Covered in `TestHudiMergeRequiredColumns`.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiFile;
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.RunLengthEncodedBlock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import static io.trino.metastore.Partitions.makePartName;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileModifiedTimeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileSizeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPartitionColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPathColumnHandle;
+import static io.trino.plugin.hive.util.HiveUtil.getPrefilledColumnValue;
+import static io.trino.spi.type.TypeUtils.writeNativeValue;
+
+/**
+ * Per-split constant values for the output columns that are not stored in the data file: Hive-style
+ * partition columns and Trino's hidden metadata columns ({@code $path}, {@code $file_size},
+ * {@code $file_modified_time}, {@code $partition}). Value computation delegates to
+ * {@link io.trino.plugin.hive.util.HiveUtil#getPrefilledColumnValue}, the same implementation the Hive
+ * connector uses for these columns (including the {@code "\N"} hive-null partition-value convention).
+ */
+public class PrefilledColumnValues
+{
+ private final Map<String, HivePartitionKey> partitionKeysByName;
+ private final String partitionName;
+ private final String filePath;
+ private final long fileSize;
+ private final long fileModifiedTime;
+
+ public static PrefilledColumnValues create(HudiSplit hudiSplit)
+ {
+ return new PrefilledColumnValues(hudiSplit);
+ }
+
+ private PrefilledColumnValues(HudiSplit hudiSplit)
+ {
+ List<HivePartitionKey> partitionKeys = hudiSplit.getPartitionKeys();
+ // ImmutableMap preserves insertion order, so $partition renders the keys in the split's
+ // partition-column order.
+ ImmutableMap.Builder<String, HivePartitionKey> byName = ImmutableMap.builder();
+ partitionKeys.forEach(partitionKey -> byName.put(partitionKey.name(), partitionKey));
+ this.partitionKeysByName = byName.buildOrThrow();
+ this.partitionName = makePartName(
+ partitionKeys.stream().map(HivePartitionKey::name).toList(),
+ partitionKeys.stream().map(HivePartitionKey::value).toList());
+ // Parquet files will be prioritised over log files
+ HudiFile hudiFile = hudiSplit.getBaseFile().isPresent()
+ ? hudiSplit.getBaseFile().get()
+ : hudiSplit.getLogFiles().getFirst();
+ this.filePath = hudiFile.getPath();
+ this.fileSize = hudiFile.getFileSize();
+ this.fileModifiedTime = hudiFile.getModificationTime();
+ }
+
+ /**
+ * Returns whether this split can provide a value for the column, i.e. it is a partition column of
+ * the split or a hidden metadata column Hudi populates.
+ */
+ public boolean isPrefilled(HiveColumnHandle columnHandle)
+ {
+ return partitionKeysByName.containsKey(columnHandle.getName())
+ || isPathColumnHandle(columnHandle)
+ || isFileSizeColumnHandle(columnHandle)
+ || isFileModifiedTimeColumnHandle(columnHandle)
+ || isPartitionColumnHandle(columnHandle);
+ }
+
+ /**
+ * Appends the column's value for this split to the builder; appends null for a column this split
+ * cannot provide.
+ */
+ public void appendTo(HiveColumnHandle columnHandle, BlockBuilder blockBuilder)
+ {
+ writeNativeValue(columnHandle.getType(), blockBuilder, nativeValueOf(columnHandle));
+ }
+
+ /**
+ * Builds a run-length-encoded {@link Block} repeating the column's constant value for
+ * {@code positionCount} positions (null-filled for a column this split cannot provide).
+ */
+ public Block toRleBlock(HiveColumnHandle columnHandle, int positionCount)
+ {
+ return RunLengthEncodedBlock.create(columnHandle.getType(), nativeValueOf(columnHandle), positionCount);
+ }
+
+ private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ if (!isPrefilled(columnHandle)) {
+ // Lenient null fill, e.g. for a hidden column Trino defines but Hudi does not populate.
+ return null;
+ }
+ HivePartitionKey partitionKey = partitionKeysByName.get(columnHandle.getName());
+ return getPrefilledColumnValue(
Review Comment:
UTC was the accident -- the old code diverged from what the Hive connector renders for `$file_modified_time`. Added it to the delta list in the commit message and noted it in the class javadoc.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiNonProjectionCompatibleMerger.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer;
+import io.trino.plugin.hudi.testing.NonProjectionCompatibleRankMerger;
+import io.trino.testing.AbstractTestQueryFramework;
+import io.trino.testing.QueryRunner;
+import org.junit.jupiter.api.Test;
+
+import static io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.RT_TABLE_NAME;
+import static io.trino.plugin.hudi.testing.NonProjectionCompatibleMergerHudiTablesInitializer.TABLE_NAME;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Acceptance test for full-table-schema merge reads (apache/hudi#19249, issue comment on scope):
+ * {@link NonProjectionCompatibleRankMerger} does NOT override {@code isProjectionCompatible()} (default
+ * {@code false}) and does NOT declare {@code merge_rank} mandatory, so the file-group reader demands the
+ * FULL table schema as its required schema for base and log reads alike, and nothing prepends
+ * {@code merge_rank} into the connector's read projection.
+ * <p>
+ * The queries below never project {@code merge_rank}. The data is laid out so each merge direction is
+ * proven independently: {@code k1}'s winning rank is on the LOG record (update wins, value 99) and
+ * {@code k2}'s winning rank is on the BASE record (base wins, value 100). A correct result therefore
+ * requires the un-projected rank column to be read on BOTH sides of the merge. {@code sum(value)} is a
+ * three-way discriminator: merged = 199, base-only = 110, built-in newest-wins = 103.
+ */
+public class TestHudiNonProjectionCompatibleMerger
Review Comment:
Added `payload_only_mor`: a table version 6 MOR table whose `hoodie.properties` says nothing about merging beyond `hoodie.compaction.payload.class` (a rank-based test payload). Two quirks worth knowing: `TableBuilder` insists on persisting inferred merge mode/strategy id for pre-v9 tables, so the initializer strips them and asserts what's left; and editing `hoodie.properties` out-of-band stales Hadoop's `.crc` sidecar, which has to be deleted. Three new cases in `TestHudiNonProjectionCompatibleMerger` (RO, row-level RT, and narrow-projection `sum(value) = 199`), all with merger impls cleared so resolution comes purely from table config.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/NonProjectionCompatibleMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,242 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Column;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.metastore.PrincipalPrivileges;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.plugin.hudi.HudiConnector;
+import io.trino.spi.security.ConnectorIdentity;
+import io.trino.testing.QueryRunner;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_REALTIME_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS;
+import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS;
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at test runtime configured with the
+ * NON-projection-compatible {@link NonProjectionCompatibleRankMerger} via {@link RecordMergeMode#CUSTOM}.
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of the same keys (log files), with
+ * inline compaction disabled so the merger runs at read time over a full-table-schema read.
+ * <p>
+ * Data is laid out so each merge direction is distinguishable: one key's winning rank comes from the LOG
+ * record and the other's from the BASE record, so a correct result proves both sides of the merge supplied
+ * the (never projected) {@code merge_rank} column. See {@code TestHudiNonProjectionCompatibleMerger}.
+ */
+public class NonProjectionCompatibleMergerHudiTablesInitializer
Review Comment:
Did it in this PR since the payload-only fixture above would have been a fourth copy: `AbstractMergerHudiTablesInitializer` owns the template and scaffolding, and the three initializers keep only their actual deltas.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3664287984
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +624,94 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} keeps only metastore data
+ * columns, and the merge read path recovers any name the metastore does not carry from the resolved
+ * table schema ({@link #appendMissingMergeRequiredColumns}). For those two fields that mirrors the
+ * file-group reader's own schema gate, so tables whose schema has neither read nothing extra; the
+ * ordering, record-key and delete-key names the reader adds without a schema check, but they are
+ * table columns whenever the table config is coherent.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
+ requiredColumnNames.add(OPERATION_METADATA_FIELD);
+ String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
+ String deleteMarker = tableConfig.getProps().getProperty(DELETE_MARKER);
+ // DeleteContext only honors a custom delete key when the marker value is also set
+ if (!StringUtils.isNullOrEmpty(deleteKey) && !StringUtils.isNullOrEmpty(deleteMarker)) {
+ requiredColumnNames.add(deleteKey);
+ }
+ return requiredColumnNames;
+ }
+
+ /**
+ * Returns {@code projection} extended with handles for the merge-required column names it does not
+ * already carry (matched case-insensitively) and the metastore could not resolve, typed from
+ * {@code dataSchema} via {@link #toColumnHandle}: the table-config-derived names
+ * ({@link #mergeRequiredColumnNames}) plus, under a CUSTOM merge mode, the mandatory fields declared
+ * by the same resolved merger the file-group reader will use. Names absent from the schema as well are
+ * dropped -- for {@code _hoodie_is_deleted} and {@code _hoodie_operation} that mirrors the file-group
+ * reader's own schema gate; the rest are table columns whenever the table config is coherent. A
+ * metastore that omits a field the table schema carries -- hive sync with
+ * {@code omit_metadata_fields=true} drops {@code _hoodie_operation}, hand-written external DDL can drop
+ * any data column -- must not starve the base-file read of it; the
+ * {@code HudiTrinoReaderContext.getFileRecordIterator} guard would otherwise fail the read. Called on
+ * the merge read path only, where the table schema is already resolved, so the recovery costs no I/O.
+ */
+ public static List<HiveColumnHandle> appendMissingMergeRequiredColumns(
+ HoodieSchema dataSchema,
+ List<HiveColumnHandle> projection,
+ HoodieTableConfig tableConfig,
+ TypedProperties readerProps)
+ {
+ Pair<RecordMergeMode, String> mergeModeAndStrategyId = resolveMergeModeAndStrategyId(tableConfig);
+ LinkedHashSet<String> requiredNames = mergeRequiredColumnNames(tableConfig, mergeModeAndStrategyId.getLeft());
+ // A CUSTOM merger can declare mandatory fields of its own (getMandatoryFieldsForMerging); ask the
+ // same resolved merger the file-group reader will use, against the same resolved schema, so its
+ // declarations are recovered too. Resolution is classloading only -- no I/O.
+ if (mergeModeAndStrategyId.getLeft() == RecordMergeMode.CUSTOM && !StringUtils.isNullOrEmpty(mergeModeAndStrategyId.getRight())) {
+ String mergeImplClasses = readerProps.getString(RECORD_MERGE_IMPL_CLASSES_WRITE_CONFIG_KEY,
+ readerProps.getString(RECORD_MERGE_IMPL_CLASSES_DEPRECATED_WRITE_CONFIG_KEY, ""));
+ Option<HoodieRecordMerger> merger = HoodieRecordUtils.createValidRecordMerger(EngineType.JAVA, mergeImplClasses, mergeModeAndStrategyId.getRight());
+ if (merger.isPresent()) {
Review Comment:
Added both next to the other `appendMissingMergeRequiredColumns` tests, and mutation-checked them: dropping the null-id half of the condition fails with the original NPE, bypassing `isPresent()` fails with `NoSuchElementException`. One trap worth flagging: the first version of the unresolvable-merger test used the all-zeros uuid, which turns out to be `PAYLOAD_BASED_MERGE_STRATEGY_UUID` and always resolves `HoodieAvroRecordMerger` -- the test passed vacuously until the mutation check caught it. It now uses an id nothing declares, with a comment warning about the collision.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous merged PR #19288:
URL: https://github.com/apache/hudi/pull/19288
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5090728695
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 74e1c977c82f3f88867bd5fdc9774d4366592e57 Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566)
* 6569ea38b806659ffd0e192334fa2f4dd4609efc 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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5091044686
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6569ea38b806659ffd0e192334fa2f4dd4609efc Azure: [FAILURE](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5093066440
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 65f09d7f048606cb5f625bdec33492c7362effda Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735)
<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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662221275
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +602,38 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} drops any name that is not a
+ * data column of the table, so tables without them read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
Review Comment:
A CUSTOM merger's own mandatory fields are still resolved metastore-only, so a merger-declared column the metastore lacks (`MaxRankRecordMerger`'s `merge_rank`) still starves the base read and trips the `getFileRecordIterator` guard. The call site already holds `readerProps` and the resolved `dataSchema`, so asking the merger there too would close it with no extra I/O - or is that out of scope here?
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiFile;
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.RunLengthEncodedBlock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import static io.trino.metastore.Partitions.makePartName;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileModifiedTimeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileSizeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPartitionColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPathColumnHandle;
+import static io.trino.plugin.hive.util.HiveUtil.getPrefilledColumnValue;
+import static io.trino.spi.type.TypeUtils.writeNativeValue;
+
+/**
+ * Per-split constant values for the output columns that are not stored in the data file: Hive-style
+ * partition columns and Trino's hidden metadata columns ({@code $path}, {@code $file_size},
+ * {@code $file_modified_time}, {@code $partition}). Value computation delegates to
+ * {@link io.trino.plugin.hive.util.HiveUtil#getPrefilledColumnValue}, the same implementation the Hive
+ * connector uses for these columns (including the {@code "\N"} hive-null partition-value convention).
+ */
+public class PrefilledColumnValues
+{
+ private final Map<String, HivePartitionKey> partitionKeysByName;
+ private final String partitionName;
+ private final String filePath;
+ private final long fileSize;
+ private final long fileModifiedTime;
+
+ public static PrefilledColumnValues create(HudiSplit hudiSplit)
+ {
+ return new PrefilledColumnValues(hudiSplit);
+ }
+
+ private PrefilledColumnValues(HudiSplit hudiSplit)
+ {
+ List<HivePartitionKey> partitionKeys = hudiSplit.getPartitionKeys();
+ // ImmutableMap preserves insertion order, so $partition renders the keys in the split's
+ // partition-column order.
+ ImmutableMap.Builder<String, HivePartitionKey> byName = ImmutableMap.builder();
+ partitionKeys.forEach(partitionKey -> byName.put(partitionKey.name(), partitionKey));
+ this.partitionKeysByName = byName.buildOrThrow();
+ this.partitionName = makePartName(
+ partitionKeys.stream().map(HivePartitionKey::name).toList(),
+ partitionKeys.stream().map(HivePartitionKey::value).toList());
+ // Parquet files will be prioritised over log files
+ HudiFile hudiFile = hudiSplit.getBaseFile().isPresent()
+ ? hudiSplit.getBaseFile().get()
+ : hudiSplit.getLogFiles().getFirst();
+ this.filePath = hudiFile.getPath();
+ this.fileSize = hudiFile.getFileSize();
+ this.fileModifiedTime = hudiFile.getModificationTime();
+ }
+
+ /**
+ * Returns whether this split can provide a value for the column, i.e. it is a partition column of
+ * the split or a hidden metadata column Hudi populates.
+ */
+ public boolean isPrefilled(HiveColumnHandle columnHandle)
+ {
+ return partitionKeysByName.containsKey(columnHandle.getName())
+ || isPathColumnHandle(columnHandle)
+ || isFileSizeColumnHandle(columnHandle)
+ || isFileModifiedTimeColumnHandle(columnHandle)
+ || isPartitionColumnHandle(columnHandle);
+ }
+
+ /**
+ * Appends the column's value for this split to the builder; appends null for a column this split
+ * cannot provide.
+ */
+ public void appendTo(HiveColumnHandle columnHandle, BlockBuilder blockBuilder)
+ {
+ writeNativeValue(columnHandle.getType(), blockBuilder, nativeValueOf(columnHandle));
+ }
+
+ /**
+ * Builds a run-length-encoded {@link Block} repeating the column's constant value for
+ * {@code positionCount} positions (null-filled for a column this split cannot provide).
+ */
+ public Block toRleBlock(HiveColumnHandle columnHandle, int positionCount)
+ {
+ return RunLengthEncodedBlock.create(columnHandle.getType(), nativeValueOf(columnHandle), positionCount);
+ }
+
+ private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ if (!isPrefilled(columnHandle)) {
+ // Lenient null fill, e.g. for a hidden column Trino defines but Hudi does not populate.
+ return null;
+ }
+ HivePartitionKey partitionKey = partitionKeysByName.get(columnHandle.getName());
+ return getPrefilledColumnValue(
Review Comment:
The class javadoc note landed, but the Summary and Changelog still lists three deltas without the zone change, and the round-2 behaviours are missing from it too - a CUSTOM table with no persisted strategy id now fails with `HUDI_BAD_DATA`, and under `hudi.parquet.use-column-names=false` a column absent from the base file now null-fills instead of erroring. Worth folding all of them in, since that text becomes the squashed commit message.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "voonhous (via GitHub)" <gi...@apache.org>.
voonhous commented on code in PR #19288:
URL: https://github.com/apache/hudi/pull/19288#discussion_r3662903760
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +602,38 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} drops any name that is not a
+ * data column of the table, so tables without them read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
Review Comment:
Not out of scope -- done. `appendMissingMergeRequiredColumns` now takes `readerProps` and, under CUSTOM mode, asks the same resolved merger for `getMandatoryFieldsForMerging` against the resolved schema, so merger-declared columns are recovered too. Pinned both ways: a unit test with `MaxRankRecordMerger`, and a fourth fixture (`omitted_rank_field_mor`) whose metastore omits `merge_rank` -- its tests fail with the guard error if the merger-ask is removed.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -348,6 +365,135 @@ public static Schema.Field getFieldFromSchema(String columnName, Schema schema)
"Failed to get column " + columnName + " from table schema");
}
+ /**
+ * Builds a {@link HiveColumnHandle} for a table-schema field, typing it from the field's Avro schema.
+ * Used to resolve columns the file-group reader requires for merging but the connector projection does
+ * not carry (e.g. {@code _hoodie_commit_time} or a delete-marker column on a narrow query).
+ * <p>
+ * The handle's {@code hiveColumnIndex} is a placeholder: page sources built from these handles resolve
+ * parquet columns by NAME (directly when {@code hudi.parquet.use-column-names=true}, otherwise
+ * {@link HudiPageSourceProvider#remapColumnIndicesToPhysical} rebuilds every index from the file
+ * schema by name), so the ordinal is never used to locate data.
+ * <p>
+ * Avro types whose Trino mapping has no Hive counterpart (uuid, time-millis/micros) fail with
+ * NOT_SUPPORTED from {@link HiveTypeTranslator#toHiveType}; such columns are equally unreadable
+ * through the metastore schema, so this surfaces the same limitation with a clear error.
+ */
+ public static HiveColumnHandle toColumnHandle(HoodieSchemaField field)
+ {
+ Type trinoType;
+ try {
+ trinoType = AVRO_TYPE_HANDLER.typeFor(field.schema().toAvroSchema());
+ }
+ catch (AvroTypeException e) {
+ throw new TrinoException(HUDI_SCHEMA_ERROR,
+ "Failed to map Avro type of column " + field.name() + " to a Trino type", e);
+ }
+ return new HiveColumnHandle(
+ field.name(),
+ 0,
+ HiveTypeTranslator.toHiveType(trinoType),
+ trinoType,
+ Optional.empty(),
+ HiveColumnHandle.ColumnType.REGULAR,
+ Optional.empty());
+ }
+
+ /**
+ * Resolves the merge mode and merge strategy id the file-group reader will use for this table,
+ * applying hudi-common's version-gated inference: the merge MODE is inferred for any table below
+ * version 9 ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}), while the STRATEGY ID
+ * the merger is resolved with is inferred only below version 8
+ * ({@code HoodieReaderContext.initRecordMerger}). The asymmetry is deliberate and must mirror
+ * hudi-common exactly: pre-v9 tables may persist neither config (a 0.x table with a custom payload
+ * class infers CUSTOM mode with the payload-based strategy id), so a connector-side decision built
+ * on the raw configs would disagree with the file-group reader's.
+ */
+ public static Pair<RecordMergeMode, String> resolveMergeModeAndStrategyId(HoodieTableConfig tableConfig)
+ {
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode();
+ String mergeStrategyId = tableConfig.getRecordMergeStrategyId();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.NINE)) {
+ Triple<RecordMergeMode, String, String> inferred = HoodieTableConfig.inferMergingConfigsForPreV9Table(
+ tableConfig.getRecordMergeMode(),
+ tableConfig.getPayloadClass(),
+ tableConfig.getRecordMergeStrategyId(),
+ tableConfig.getOrderingFieldsStr().orElse(null),
+ tableConfig.getTableVersion());
+ mergeMode = inferred.getLeft();
+ if (tableConfig.getTableVersion().lesserThan(HoodieTableVersion.EIGHT)) {
+ mergeStrategyId = inferred.getRight();
+ }
+ }
+ return Pair.of(mergeMode, mergeStrategyId);
+ }
+
+ /**
+ * Returns whether reads of this table go through a CUSTOM record merger that is NOT projection
+ * compatible. For such mergers the file-group reader demands the FULL table schema as its required
+ * schema on any split with log files ({@code FileGroupReaderSchemaHandler.generateRequiredSchema}),
+ * so the connector must read every table column, not just the query projection.
+ * <p>
+ * Callers must pass the merge mode and strategy id resolved by {@link #resolveMergeModeAndStrategyId}
+ * so the {@link HoodieRecordUtils#createValidRecordMerger} call here sees the same inputs as the
+ * file-group reader's own resolution. A CUSTOM mode without a strategy id returns false: no projection
+ * expansion is possible without a merger to ask, and {@code HudiTrinoReaderContext.getRecordMerger}
+ * rejects the read with an actionable error when the merger is actually needed -- hudi-common's
+ * {@link HoodieRecordUtils#createValidRecordMerger} would otherwise NPE on the null id.
+ * <p>
+ * Note the payload-based strategy id resolves {@code HoodieAvroRecordMerger}, which keeps the
+ * interface default {@code isProjectionCompatible() == false} -- so EVERY custom-payload table takes
+ * the full-schema read path, not just tables with exotic custom mergers. That matches the file-group
+ * reader (and Spark); the extra I/O is inherent to payload-based merging.
+ */
+ public static boolean usesNonProjectionCompatibleMerger(RecordMergeMode mergeMode, String mergeStrategyId, String mergeImplClasses)
+ {
+ if (mergeMode != RecordMergeMode.CUSTOM || StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ return false;
+ }
+ Option<HoodieRecordMerger> merger = HoodieRecordUtils.createValidRecordMerger(EngineType.JAVA, mergeImplClasses, mergeStrategyId);
+ return merger.isPresent() && !merger.get().isProjectionCompatible();
+ }
+
+ /**
+ * Rejects a CUSTOM-mode read that has no merge strategy id to resolve a record merger with. The
+ * strategy id is only inferred below table version 8 ({@link #resolveMergeModeAndStrategyId}), so a
+ * version 8 table written without a persisted id reaches merger resolution with a null id, which
+ * {@link HoodieRecordUtils#createValidRecordMerger} dereferences straight away.
+ */
+ public static void validateCustomMergeStrategyId(String mergeStrategyId)
+ {
+ if (StringUtils.isNullOrEmpty(mergeStrategyId)) {
+ String strategyIdKey = HoodieTableConfig.RECORD_MERGE_STRATEGY_ID.key();
+ throw new TrinoException(HUDI_BAD_DATA,
+ ("Table resolved to %s merge mode but persists no `%s`, so no record merger can be resolved for merging log files. "
+ + "Table version 8 tables written without a persisted strategy id resolve this way; set `%s` on the table to the "
Review Comment:
Widened both to "version 8 or later".
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/HudiUtil.java:
##########
@@ -471,12 +622,73 @@ public static List<HiveColumnHandle> getMergeRequiredColumnHandles(
}
}
- if (requiredColumnNames.isEmpty()) {
- return Collections.emptyList();
- }
return buildColumnHandles(table, typeManager, requiredColumnNames, timestampPrecision);
}
+ /**
+ * The merge-mandatory column names that do not depend on a custom merger, mirroring the non-CUSTOM
+ * branch of {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging}. The delete-marker and
+ * operation fields are added unconditionally: {@code buildColumnHandles} keeps only metastore data
+ * columns, and the merge read path recovers any name the metastore does not carry from the resolved
+ * table schema ({@link #appendMissingMergeRequiredColumns}) -- the gate the file-group reader itself
+ * applies -- so tables whose schema has neither field read nothing extra.
+ */
+ @VisibleForTesting
+ static LinkedHashSet<String> mergeRequiredColumnNames(HoodieTableConfig tableConfig, RecordMergeMode recordMergeMode)
+ {
+ LinkedHashSet<String> requiredColumnNames = new LinkedHashSet<>();
+ if (recordMergeMode != null && recordMergeMode != RecordMergeMode.COMMIT_TIME_ORDERING) {
+ requiredColumnNames.addAll(tableConfig.getOrderingFields());
+ }
+ // Without populated meta fields the file-group reader keys records on the record-key data columns
+ if (!tableConfig.populateMetaFields()) {
+ tableConfig.getRecordKeyFields().ifPresent(fields -> requiredColumnNames.addAll(Arrays.asList(fields)));
+ }
+ // Delete markers and the operation field decide record deletion at merge time
+ requiredColumnNames.add(HOODIE_IS_DELETED_FIELD);
+ requiredColumnNames.add(OPERATION_METADATA_FIELD);
+ String deleteKey = tableConfig.getProps().getProperty(DELETE_KEY);
+ String deleteMarker = tableConfig.getProps().getProperty(DELETE_MARKER);
+ // DeleteContext only honors a custom delete key when the marker value is also set
+ if (!StringUtils.isNullOrEmpty(deleteKey) && !StringUtils.isNullOrEmpty(deleteMarker)) {
+ requiredColumnNames.add(deleteKey);
+ }
+ return requiredColumnNames;
+ }
+
+ /**
+ * Returns {@code projection} extended with handles for the merge-required column names it does not
+ * already carry (matched case-insensitively), typed from {@code dataSchema} via {@link #toColumnHandle};
+ * names absent from the schema as well are dropped. The Avro table schema is the gate the file-group
Review Comment:
Scoped it: the javadoc now says the schema gate is the reader's own only for `_hoodie_is_deleted`/`_hoodie_operation`, and that the ordering/record-key/delete-key names are added without a schema check but are table columns whenever the config is coherent.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMergeRequiredColumns.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi;
+
+import io.trino.metastore.HiveType;
+import io.trino.plugin.hive.HiveColumnHandle;
+import org.apache.avro.SchemaBuilder;
+import org.apache.hudi.common.config.RecordMergeMode;
+import org.apache.hudi.common.schema.HoodieSchema;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Optional;
+
+import static io.trino.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.trino.plugin.hive.HiveColumnHandle.createBaseColumn;
+import static io.trino.plugin.hudi.HudiUtil.appendMissingMergeRequiredColumns;
+import static io.trino.plugin.hudi.HudiUtil.mergeRequiredColumnNames;
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_KEY;
+import static org.apache.hudi.common.model.DefaultHoodieRecordPayload.DELETE_MARKER;
+import static org.apache.hudi.common.model.HoodieRecord.HOODIE_IS_DELETED_FIELD;
+import static org.apache.hudi.common.model.HoodieRecord.OPERATION_METADATA_FIELD;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests {@link HudiUtil#mergeRequiredColumnNames}, which mirrors the non-CUSTOM branch of the file-group
+ * reader's {@code FileGroupReaderSchemaHandler.getMandatoryFieldsForMerging} so the base-file read
+ * projection carries every column the merge may consult (ordering fields, delete markers, the operation
+ * field, record-key data columns) even when the query does not project them, plus the merge-path recovery
+ * of names the metastore could not resolve ({@link HudiUtil#appendMissingMergeRequiredColumns}).
+ */
+class TestHudiMergeRequiredColumns
+{
+ @Test
+ public void testOrderingFieldsFollowMergeMode()
+ {
+ HoodieTableConfig tableConfig = new HoodieTableConfig();
+ tableConfig.setValue(HoodieTableConfig.ORDERING_FIELDS, "ts,seq");
+
+ assertThat(mergeRequiredColumnNames(tableConfig, RecordMergeMode.EVENT_TIME_ORDERING))
+ .contains("ts", "seq");
+ // Commit-time merging ignores ordering fields
+ assertThat(mergeRequiredColumnNames(tableConfig, RecordMergeMode.COMMIT_TIME_ORDERING))
+ .doesNotContain("ts", "seq");
+ assertThat(mergeRequiredColumnNames(tableConfig, null))
+ .doesNotContain("ts", "seq");
+ }
+
+ @Test
+ public void testDeleteAndOperationColumnsAlwaysRequested()
+ {
+ // Requested unconditionally: getMergeRequiredColumnHandles drops names that are in neither the
Review Comment:
Reworded to split the claim: the metastore half in `getMergeRequiredColumnHandles`, the schema half in `appendMissingMergeRequiredColumns` on the page-source path.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/AbstractMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,291 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.trino.filesystem.Location;
+import io.trino.filesystem.TrinoFileSystem;
+import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.metastore.Column;
+import io.trino.metastore.HiveMetastore;
+import io.trino.metastore.HiveMetastoreFactory;
+import io.trino.metastore.PrincipalPrivileges;
+import io.trino.metastore.StorageFormat;
+import io.trino.metastore.Table;
+import io.trino.plugin.hudi.HudiConnector;
+import io.trino.spi.security.ConnectorIdentity;
+import io.trino.testing.QueryRunner;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.common.HoodieJavaEngineContext;
+import org.apache.hudi.common.bootstrap.index.NoOpBootstrapIndex;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieAvroRecord;
+import org.apache.hudi.common.model.HoodieKey;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.marker.MarkerType;
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieIndexConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.index.HoodieIndex;
+import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static com.google.common.io.MoreFiles.deleteRecursively;
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.HUDI_PARQUET_REALTIME_INPUT_FORMAT;
+import static io.trino.hive.formats.HiveClassNames.MAPRED_PARQUET_OUTPUT_FORMAT_CLASS;
+import static io.trino.hive.formats.HiveClassNames.PARQUET_HIVE_SERDE_CLASS;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE;
+import static java.nio.file.Files.createTempDirectory;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Shared machinery for the non-partitioned Merge-On-Read fixtures that exist to exercise record merging at
+ * read time. The table is written by the Hudi Java write client into a local staging directory and then
+ * mirrored into the Trino filesystem the connector reads from; two metastore tables are registered against
+ * that location, a read-optimized one (base files only) and a real-time one (suffix {@code _rt}) that merges
+ * base + log files through the file group reader.
+ * <p>
+ * Inline compaction is off for every fixture so the log files written by the delta commits survive and must
+ * be merged at read time, and the metadata table is off because MDT writes need hbase dependencies that are
+ * not on the Trino test classpath.
+ * <p>
+ * Subclasses supply the schema, the merge-related table and write configuration, and the commits; everything
+ * a fixture does not vary lives here.
+ */
+public abstract class AbstractMergerHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ /** Every fixture built on this base is keyed on {@code key} and ordered by {@code ts}, in a single unnamed partition. */
+ protected static final String RECORD_KEY_FIELD = "key";
+ protected static final String ORDERING_FIELD = "ts";
+
+ private static final String PARTITION_PATH = "";
+
+ /** Hudi metadata columns, prepended to every table's data columns in the metastore definition. */
+ private static final List<Column> HUDI_META_COLUMNS = ImmutableList.of(
+ new Column("_hoodie_commit_time", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_commit_seqno", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_record_key", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_partition_path", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("_hoodie_file_name", HIVE_STRING, Optional.empty(), Map.of()));
+
+ private final String tableName;
+
+ private TrinoFileSystem fileSystem;
+ private Location tableLocation;
+ private java.nio.file.Path stagingDir;
+ private Path stagingTablePath;
+ private HoodieJavaWriteClient<HoodieAvroPayload> writeClient;
+
+ protected AbstractMergerHudiTablesInitializer(String tableName)
+ {
+ this.tableName = requireNonNull(tableName, "tableName is null");
+ }
+
+ @Override
+ public final void initializeTables(QueryRunner queryRunner, Location externalLocation, String schemaName)
+ throws Exception
+ {
+ fileSystem = ((HudiConnector) queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(TrinoFileSystemFactory.class)
+ .create(ConnectorIdentity.ofUser("test"));
+ HiveMetastore metastore = ((HudiConnector) queryRunner.getCoordinator().getConnector("hudi")).getInjector()
+ .getInstance(HiveMetastoreFactory.class)
+ .createMetastore(Optional.empty());
+
+ tableLocation = externalLocation.appendPath(tableName);
+ stagingDir = createTempDirectory(tableName.replace('_', '-'));
+ stagingTablePath = new Path(stagingDir.resolve(tableName).toUri());
+
+ boolean initialized = false;
+ try {
+ initTable();
+ afterTableInit();
+ writeClient = createWriteClient();
+ writeInitialCommits(writeClient);
+ syncToTrino();
+
+ metastore.createTable(createTableDefinition(schemaName, tableName, false), PrincipalPrivileges.NO_PRIVILEGES);
+ metastore.createTable(createTableDefinition(schemaName, tableName + "_rt", true), PrincipalPrivileges.NO_PRIVILEGES);
+ initialized = true;
+ }
+ finally {
+ // Only fixtures that keep writing commits after initialization need the staging directory and the
+ // write client to outlive this call; they are responsible for calling close().
+ if (!initialized || !keepsWriterOpen()) {
+ close();
+ }
+ }
+ }
+
+ /** The table's data columns, in schema order; the Hudi metadata columns are prepended by this class. */
+ protected abstract List<Column> dataColumns();
+
+ /** The Avro schema the write client writes, matching {@link #dataColumns()}. */
+ protected abstract Schema avroSchema();
+
+ /**
+ * Applies the fixture's merge-related table configuration (merge mode, merge strategy id, payload class).
+ * The table type, record key fields and ordering fields are set by this class.
+ */
+ protected abstract void configureTableConfig(HoodieTableMetaClient.TableBuilder tableBuilder);
+
+ /** Applies the fixture's merge-related write configuration (merge mode, merge strategy id, merger impl classes). */
+ protected abstract void configureWriteConfig(HoodieWriteConfig.Builder writeConfigBuilder);
+
+ /** Writes the commits that make up the fixture's initial state. */
+ protected abstract void writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException;
+
+ /** Runs on the freshly created table, before the write client exists and before any data is written. */
+ protected void afterTableInit()
+ throws IOException {}
+
+ /** Inline compaction is disabled, so this only has to stay above the number of delta commits a fixture writes. */
+ protected int maxDeltaCommitsBeforeCompaction()
+ {
+ return 100;
+ }
+
+ /** Whether the staging directory and write client survive {@link #initializeTables}; such fixtures must call {@link #close()}. */
+ protected boolean keepsWriterOpen()
+ {
+ return false;
+ }
+
+ public void close()
+ throws IOException
+ {
+ if (writeClient != null) {
+ writeClient.close();
+ writeClient = null;
+ }
+ if (stagingDir != null) {
+ deleteRecursively(stagingDir, ALLOW_INSECURE);
+ stagingDir = null;
+ }
+ }
+
+ /** Local directory the write client writes into, before {@link #syncToTrino()} mirrors it to the connector. */
+ protected java.nio.file.Path stagingTableDirectory()
+ {
+ return stagingDir.resolve(tableName);
+ }
+
+ protected HoodieJavaWriteClient<HoodieAvroPayload> writeClient()
+ {
+ return writeClient;
+ }
+
+ protected static HoodieRecord<HoodieAvroPayload> avroRecord(GenericRecord record, String key)
+ {
+ return new HoodieAvroRecord<>(new HoodieKey(key, PARTITION_PATH), new HoodieAvroPayload(Option.of(record)), null);
+ }
+
+ /** Mirrors the staged table into the Trino filesystem so the connector observes the commits written so far. */
+ protected void syncToTrino()
+ {
+ try {
+ if (fileSystem.directoryExists(tableLocation).orElse(false)) {
+ fileSystem.deleteDirectory(tableLocation);
+ }
+ ResourceHudiTablesInitializer.copyDir(stagingTableDirectory(), fileSystem, tableLocation);
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Failed to sync staged Hudi table to Trino filesystem", e);
+ }
+ }
+
+ private void initTable()
+ {
+ HoodieTableMetaClient.TableBuilder tableBuilder = HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.MERGE_ON_READ)
+ .setTableName(tableName)
+ .setTimelineLayoutVersion(1)
+ .setBootstrapIndexClass(NoOpBootstrapIndex.class.getName())
+ .setRecordKeyFields(RECORD_KEY_FIELD)
+ .setOrderingFields(ORDERING_FIELD);
+ configureTableConfig(tableBuilder);
+ try {
+ tableBuilder.initTable(new HadoopStorageConfiguration(new Configuration()), stagingTablePath.toString());
+ }
+ catch (IOException e) {
+ throw new RuntimeException("Could not init table " + tableName, e);
+ }
+ }
+
+ private HoodieJavaWriteClient<HoodieAvroPayload> createWriteClient()
+ {
+ Configuration conf = new Configuration();
+ HoodieWriteConfig.Builder writeConfigBuilder = HoodieWriteConfig.newBuilder()
+ .withPath(stagingTablePath.toString())
+ .withSchema(avroSchema().toString())
+ .withParallelism(2, 2)
+ .withDeleteParallelism(2)
+ .forTable(tableName)
+ .withIndexConfig(HoodieIndexConfig.newBuilder().withIndexType(HoodieIndex.IndexType.INMEMORY).build())
+ // Keep log files around so merging runs at read time.
+ .withCompactionConfig(HoodieCompactionConfig.newBuilder()
+ .withInlineCompaction(false)
+ .withMaxNumDeltaCommitsBeforeCompaction(maxDeltaCommitsBeforeCompaction())
+ .build())
+ .withEmbeddedTimelineServerEnabled(false)
+ .withMarkersType(MarkerType.DIRECT.name())
+ // MDT writes require hbase deps not present in the Trino runtime.
+ .withMetadataConfig(HoodieMetadataConfig.newBuilder().enable(false).build());
+ // The ordering field is carried by the table config (setOrderingFields); the deprecated
Review Comment:
Moved it back into the chain, at the spot `withPreCombineField` would go.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/CompositeHudiTablesInitializer.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.filesystem.Location;
+import io.trino.testing.QueryRunner;
+
+import java.util.List;
+
+/**
+ * Runs several {@link HudiTablesInitializer}s against one query runner, in order, so a single test class can
+ * load more than one table fixture. Each delegate must create its own tables under a distinct name.
+ */
+public class CompositeHudiTablesInitializer
+ implements HudiTablesInitializer
+{
+ private final List<HudiTablesInitializer> delegates;
+
+ public CompositeHudiTablesInitializer(HudiTablesInitializer... delegates)
+ {
+ this.delegates = ImmutableList.copyOf(delegates);
+ }
+
+ @Override
+ public void initializeTables(QueryRunner queryRunner, Location externalLocation, String schemaName)
+ throws Exception
+ {
+ for (HudiTablesInitializer delegate : delegates) {
+ delegate.initializeTables(queryRunner, externalLocation, schemaName);
Review Comment:
Went with the javadoc: the composite now says delegates must finish their writes within `initializeTables`, and that a `keepsWriterOpen()` fixture can't be composed since the interface has no close to forward. Didn't want to grow the interface for a case nothing needs yet.
##########
hudi-trino/src/test/java/io/trino/plugin/hudi/testing/PayloadOnlyMergerHudiTablesInitializer.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.testing;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.metastore.Column;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.hudi.client.HoodieJavaWriteClient;
+import org.apache.hudi.client.WriteStatus;
+import org.apache.hudi.common.model.HoodieAvroPayload;
+import org.apache.hudi.common.model.HoodieRecord;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.config.HoodieWriteConfig;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static io.trino.metastore.HiveType.HIVE_LONG;
+import static io.trino.metastore.HiveType.HIVE_STRING;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+/**
+ * Creates a non-partitioned Merge-On-Read table at TABLE VERSION 6 whose only merge configuration is a
+ * {@link org.apache.hudi.common.model.HoodieRecordPayload} class ({@link RankBasedTestPayload}), the way a
+ * genuine pre-1.0 table looks on storage.
+ * <p>
+ * Such a table resolves at read time to {@code CUSTOM} merge mode with the payload-based merge strategy, which
+ * in turn resolves {@code HoodieAvroRecordMerger}. That merger is not projection compatible, so the file group
+ * reader demands the FULL table schema for base and log reads alike and nothing prepends the payload's decision
+ * column into the connector's read projection.
+ * <p>
+ * One {@code bulkInsert} (base files) is followed by one {@code upsert} of the same keys (log files), with data
+ * laid out so each merge direction is distinguishable: {@code k1}'s winning rank is on the LOG record and
+ * {@code k2}'s on the BASE record. See {@code TestHudiNonProjectionCompatibleMerger}.
+ */
+public class PayloadOnlyMergerHudiTablesInitializer
+ extends AbstractMergerHudiTablesInitializer
+{
+ public static final String TABLE_NAME = "payload_only_mor";
+ public static final String RT_TABLE_NAME = TABLE_NAME + "_rt";
+
+ private static final String RANK_FIELD = RankBasedTestPayload.RANK_COLUMN;
+ private static final HoodieTableVersion TABLE_VERSION = HoodieTableVersion.SIX;
+
+ public PayloadOnlyMergerHudiTablesInitializer()
+ {
+ super(TABLE_NAME);
+ }
+
+ @Override
+ protected List<Column> dataColumns()
+ {
+ return ImmutableList.of(
+ new Column(RECORD_KEY_FIELD, HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("name", HIVE_STRING, Optional.empty(), Map.of()),
+ new Column("value", HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(RANK_FIELD, HIVE_LONG, Optional.empty(), Map.of()),
+ new Column(ORDERING_FIELD, HIVE_LONG, Optional.empty(), Map.of()));
+ }
+
+ @Override
+ protected Schema avroSchema()
+ {
+ List<Schema.Field> fields = ImmutableList.of(
+ new Schema.Field(RECORD_KEY_FIELD, Schema.create(Schema.Type.STRING)),
+ new Schema.Field("name", Schema.create(Schema.Type.STRING)),
+ new Schema.Field("value", Schema.create(Schema.Type.LONG)),
+ new Schema.Field(RANK_FIELD, Schema.create(Schema.Type.LONG)),
+ new Schema.Field(ORDERING_FIELD, Schema.create(Schema.Type.LONG)));
+ return Schema.createRecord(TABLE_NAME, null, null, false, new ArrayList<>(fields));
+ }
+
+ @Override
+ protected void configureTableConfig(HoodieTableMetaClient.TableBuilder tableBuilder)
+ {
+ // Only the payload class: the merge mode and the merge strategy id must stay out of hoodie.properties
+ // so the reader has to infer them, which is what makes this a pre-1.0 payload-only table.
+ tableBuilder
+ .setTableVersion(TABLE_VERSION)
+ .setPayloadClassName(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void configureWriteConfig(HoodieWriteConfig.Builder writeConfigBuilder)
+ {
+ writeConfigBuilder
+ .withWriteTableVersion(TABLE_VERSION.versionCode())
+ // Without this the write client silently upgrades the table to the current version on the first commit.
+ .withAutoUpgradeVersion(false)
+ .withWritePayLoad(RankBasedTestPayload.class.getName());
+ }
+
+ @Override
+ protected void afterTableInit()
+ throws IOException
+ {
+ stripInferredMergeConfigs();
+ }
+
+ @Override
+ protected void writeInitialCommits(HoodieJavaWriteClient<HoodieAvroPayload> client)
+ throws IOException
+ {
+ Schema schema = avroSchema();
+ // First commit: bulk insert base records (produces base parquet files).
+ String firstCommit = client.startCommit();
+ List<WriteStatus> firstStatuses = client.bulkInsert(ImmutableList.of(
+ record(schema, "k1", "k1_base", 10L, 5L, 1L),
+ record(schema, "k2", "k2_base", 100L, 9L, 1L)), firstCommit);
+ client.commit(firstCommit, firstStatuses);
+
+ // Second commit: upserts the same keys (produces log files since inline compaction is disabled).
+ // k1 update has a HIGHER rank (7 > 5) -> the payload keeps the update (99): the LOG record's rank decides.
+ // k2 update has a LOWER rank (1 < 9) -> the payload keeps the base record (100): the BASE record's rank
+ // decides, which only works when the base read carries merge_rank despite it never being projected.
+ String secondCommit = client.startCommit();
+ List<WriteStatus> secondStatuses = client.upsert(ImmutableList.of(
+ record(schema, "k1", "k1_updated", 99L, 7L, 2L),
+ record(schema, "k2", "k2_updated", 4L, 1L, 2L)), secondCommit);
+ client.commit(secondCommit, secondStatuses);
+
+ // The commits go through the table config; re-check that the writer left the fixture payload-only.
+ stripInferredMergeConfigs();
+ }
+
+ /**
+ * Removes the merge mode and merge strategy id that table creation infers and persists for every pre-v9
+ * table, then verifies what is left. The fixture has to mimic a genuine pre-1.0 table, which persists its
+ * payload class and nothing else about merging; leaving the inferred keys in would hand the reader the
+ * answer this fixture exists to make it derive.
+ */
+ private void stripInferredMergeConfigs()
+ throws IOException
+ {
+ Path metaDirectory = stagingTableDirectory().resolve(".hoodie");
+ Path propertiesFile = metaDirectory.resolve("hoodie.properties");
+ List<String> retainedLines = Files.readAllLines(propertiesFile, UTF_8).stream()
+ .filter(line -> !isEntryFor(line, HoodieTableConfig.RECORD_MERGE_MODE.key())
Review Comment:
Checked `dropInvalidConfigs` and confirmed: the mode's config is since-version 1.0.0, so it never reaches disk for a v6 table. The filter now strips just the strategy id, the javadoc explains why, and the `checkAbsent` on the mode stays as a pin of creation's own behavior.
##########
hudi-trino/src/main/java/io/trino/plugin/hudi/util/PrefilledColumnValues.java:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed 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 io.trino.plugin.hudi.util;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.plugin.hive.HiveColumnHandle;
+import io.trino.plugin.hive.HivePartitionKey;
+import io.trino.plugin.hudi.HudiSplit;
+import io.trino.plugin.hudi.file.HudiFile;
+import io.trino.spi.block.Block;
+import io.trino.spi.block.BlockBuilder;
+import io.trino.spi.block.RunLengthEncodedBlock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
+
+import static io.trino.metastore.Partitions.makePartName;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileModifiedTimeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isFileSizeColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPartitionColumnHandle;
+import static io.trino.plugin.hive.HiveColumnHandle.isPathColumnHandle;
+import static io.trino.plugin.hive.util.HiveUtil.getPrefilledColumnValue;
+import static io.trino.spi.type.TypeUtils.writeNativeValue;
+
+/**
+ * Per-split constant values for the output columns that are not stored in the data file: Hive-style
+ * partition columns and Trino's hidden metadata columns ({@code $path}, {@code $file_size},
+ * {@code $file_modified_time}, {@code $partition}). Value computation delegates to
+ * {@link io.trino.plugin.hive.util.HiveUtil#getPrefilledColumnValue}, the same implementation the Hive
+ * connector uses for these columns (including the {@code "\N"} hive-null partition-value convention).
+ */
+public class PrefilledColumnValues
+{
+ private final Map<String, HivePartitionKey> partitionKeysByName;
+ private final String partitionName;
+ private final String filePath;
+ private final long fileSize;
+ private final long fileModifiedTime;
+
+ public static PrefilledColumnValues create(HudiSplit hudiSplit)
+ {
+ return new PrefilledColumnValues(hudiSplit);
+ }
+
+ private PrefilledColumnValues(HudiSplit hudiSplit)
+ {
+ List<HivePartitionKey> partitionKeys = hudiSplit.getPartitionKeys();
+ // ImmutableMap preserves insertion order, so $partition renders the keys in the split's
+ // partition-column order.
+ ImmutableMap.Builder<String, HivePartitionKey> byName = ImmutableMap.builder();
+ partitionKeys.forEach(partitionKey -> byName.put(partitionKey.name(), partitionKey));
+ this.partitionKeysByName = byName.buildOrThrow();
+ this.partitionName = makePartName(
+ partitionKeys.stream().map(HivePartitionKey::name).toList(),
+ partitionKeys.stream().map(HivePartitionKey::value).toList());
+ // Parquet files will be prioritised over log files
+ HudiFile hudiFile = hudiSplit.getBaseFile().isPresent()
+ ? hudiSplit.getBaseFile().get()
+ : hudiSplit.getLogFiles().getFirst();
+ this.filePath = hudiFile.getPath();
+ this.fileSize = hudiFile.getFileSize();
+ this.fileModifiedTime = hudiFile.getModificationTime();
+ }
+
+ /**
+ * Returns whether this split can provide a value for the column, i.e. it is a partition column of
+ * the split or a hidden metadata column Hudi populates.
+ */
+ public boolean isPrefilled(HiveColumnHandle columnHandle)
+ {
+ return partitionKeysByName.containsKey(columnHandle.getName())
+ || isPathColumnHandle(columnHandle)
+ || isFileSizeColumnHandle(columnHandle)
+ || isFileModifiedTimeColumnHandle(columnHandle)
+ || isPartitionColumnHandle(columnHandle);
+ }
+
+ /**
+ * Appends the column's value for this split to the builder; appends null for a column this split
+ * cannot provide.
+ */
+ public void appendTo(HiveColumnHandle columnHandle, BlockBuilder blockBuilder)
+ {
+ writeNativeValue(columnHandle.getType(), blockBuilder, nativeValueOf(columnHandle));
+ }
+
+ /**
+ * Builds a run-length-encoded {@link Block} repeating the column's constant value for
+ * {@code positionCount} positions (null-filled for a column this split cannot provide).
+ */
+ public Block toRleBlock(HiveColumnHandle columnHandle, int positionCount)
+ {
+ return RunLengthEncodedBlock.create(columnHandle.getType(), nativeValueOf(columnHandle), positionCount);
+ }
+
+ private Object nativeValueOf(HiveColumnHandle columnHandle)
+ {
+ if (!isPrefilled(columnHandle)) {
+ // Lenient null fill, e.g. for a hidden column Trino defines but Hudi does not populate.
+ return null;
+ }
+ HivePartitionKey partitionKey = partitionKeysByName.get(columnHandle.getName());
+ return getPrefilledColumnValue(
Review Comment:
Folded them all into the Summary and Changelog: the `$file_modified_time` zone change joins the `PrefilledColumnValues` delta list, and the null-fill under `use-column-names=false`, the `HUDI_BAD_DATA` guard, and the metastore-omitted recovery each have their own bullet now. Also refreshed the stale "stacked on #18837" opener while in there.
--
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] feat(trino): resolve merge-required columns from the table schema [hudi]
Posted by "hudi-bot (via GitHub)" <gi...@apache.org>.
hudi-bot commented on PR #19288:
URL: https://github.com/apache/hudi/pull/19288#issuecomment-5104055651
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15483",
"triggerID" : "50cf40647d30e797753d6b2815adfdfa4b3aedec",
"triggerType" : "PUSH"
}, {
"hash" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15566",
"triggerID" : "74e1c977c82f3f88867bd5fdc9774d4366592e57",
"triggerType" : "PUSH"
}, {
"hash" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15733",
"triggerID" : "6569ea38b806659ffd0e192334fa2f4dd4609efc",
"triggerType" : "PUSH"
}, {
"hash" : "65f09d7f048606cb5f625bdec33492c7362effda",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15735",
"triggerID" : "65f09d7f048606cb5f625bdec33492c7362effda",
"triggerType" : "PUSH"
}, {
"hash" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15741",
"triggerID" : "dd414607d8f0a64f4fab3f3c59f0df699464f484",
"triggerType" : "PUSH"
}, {
"hash" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"status" : "DELETED",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15753",
"triggerID" : "453c01c70e6fbcf866ae4aab3cadce24d8c7dc4b",
"triggerType" : "PUSH"
}, {
"hash" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15765",
"triggerID" : "b98144b32be9bf69be0e012ce5b7a49b9c5eebae",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* b98144b32be9bf69be0e012ce5b7a49b9c5eebae Azure: [SUCCESS](https://dev.azure.com/apachehudi/a1a51da7-8592-47d4-88dc-fd67bed336bb/_build/results?buildId=15765)
<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