Repository navigation
AddFiles: Parse partition keys in filepath for requirement check - #40691
Conversation
|
/Gemini review |
|
/gemini review |
There was a problem hiding this comment.
📋 Review Summary
This PR adds support for reading partition column values from Hive-style folder paths (key=value/) during AddFiles, allowing data files to omit identity-partitioned columns even when those columns are required or pinned in the destination Iceberg table. However, the PR currently contains a critical merge/rebase regression: the first commit (a3785bdc) introduced the feature, while the subsequent commits (de7f78a5 and f50f71d6) were based on master rather than the feature commit, leaving only a 1-line change in .github/trigger_files/IO_Iceberg_Integration_Tests.json in the PR diff.
🔍 General Feedback
- Git History / Missing Changes (🔴 Critical): Commit
de7f78a5(revert-doc) appears to have been created onmaster(or reverted the entire feature commit instead of just the doc change inAddFilesSchemaTransformProvider.java), so the net diff of this PR againstapache:masteris currently just.github/trigger_files/IO_Iceberg_Integration_Tests.json. Please cherry-pick or rebasea3785bdc79b5e914a1a28a5a78d677ccb6dc5436back onto the branch. - Implementation Quality (on
a3785bdc): The feature implementation acrossPartitionPaths,CollectDistinctSchemas,ReadFooterSchema,SchemaDelta,SchemaPlan, andAddFilesis well-structured, cleanly handles__HIVE_DEFAULT_PARTITION__and non-identity partition transforms, and has thorough unit and integration test coverage. - Edge Case in
SchemaDelta.classify(🟡 Medium): InSchemaDelta.classify(Table, Schema, Schema, Set<Integer>)(froma3785bdc), the early returnif (before.sameSchema(fileSchema))does not checkColumnNameChecks.problem(fileSchema), whereas a schema differing only by a path-supplied partition column falls through toColumnNameChecks.problem(fileSchema). Even More importantly, ifbeforehas a required identity-partitioned columndtandfileSchemaomitsdtwhilefilledOutsideFilecontainsdt's field ID,findAbsentRequiredskipsdt(absentis empty), butclassifystill runsunionByNameWith(fileSchema)anddiffStruct(before.asStruct(), merged.asStruct(), ...)—which works becauseunionByNameWithdoes not relax missing columns unlessmakeOptionalis called, though it still opens a scratch transaction and runsunionByNameWithunnecessarily when the only difference is path-supplied columns.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
|
/gemini review |
There was a problem hiding this comment.
📋 Review Summary
This PR introduces partition-path awareness for schema evolution, a dry-run reporting mode (DryRunReport), and Parquet field-ID conflict validation (ParquetFieldIds) for AddFiles. The overall design is well-structured and cleanly separates concerns across the pre-pass, commit plan, and per-file registration stages.
🔍 General Feedback
- There are a couple of edge cases in
ParquetFieldIdsandAddFileswhere Parquet schemas with duplicate field IDs or pinned identity-partition columns provided via path can cause unexpected failures or false rejections. - URL-encoded partition key names in Hive-style folder paths are not decoded before matching against Iceberg
PartitionSpecfield names.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
💬 Inline Findings
-
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java(line 717): 🟠 High:checkPinnedColumnsdoes not account forfilledByPartitionPath(unlikeSchemaDelta.classifyright above it). If a required/pinned identity-partition column is provided by the file's directory path (e.g.,dt=2024-01-15/f.parquet) and omitted from the Parquet file itself,SchemaPlanandSchemaDeltaaccept the file, butcheckPinnedColumnswill reject it here withmissing required columnbecausefileSchema.findField(fileColumn)returnsnull. Columns whose table field ID is infilledByPartitionPathshould be skipped incheckPinnedColumns. -
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ParquetFieldIds.java(line 122): 🟠 High:TypeUtil.indexByIdandfileSchema.findColumnName(id)can throwIllegalArgumentException(from GuavaImmutableMap.Builderduplicate key conflict when two columns in the file share the same Parquet field ID) or fail if a nested struct/list/map shares an ID with another field. Because this call is outside thetry / catch (RuntimeException e)block on lines 116-120, a file with duplicate field IDs will throw an unhandledIllegalArgumentExceptioninstead of aConflictException, crashing the bundle rather than routing the file to the error output. -
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionPaths.java(line 62): 🟡 Medium: Hive-style partition paths URL-encode special characters in both keys and values (andTableUtils.parsePartitionPathusesURLDecoder.decode(part, "UTF-8")). Extractingparts[0]and comparingHIVE_NULL.equals(parts[1])without URL-decoding means partition field names containing special characters (like spaces or colons) won't matchPartitionField.name()incolumnsFilledByKeys.
|
R: @ahmedabu98 |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40691 +/- ##
=============================================
- Coverage 56.16% 0 -56.17%
=============================================
Files 1124 0 -1124
Lines 178336 0 -178336
Branches 1489 0 -1489
=============================================
- Hits 100170 0 -100170
+ Misses 75645 0 -75645
+ Partials 2521 0 -2521
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
ahmedabu98
left a comment
There was a problem hiding this comment.
thanks, this LGTM!
just a nit regarding naming to avoid confusion for future readers/agents
also maybe should have an end-to-end test with /dt=__HIVE_DEFAULT_PARTITION__/
ahmedabu98
left a comment
There was a problem hiding this comment.
Ahh sorry looks like my comments didnt get thru last time
With
schema_evolution_optionsandlocation_prefixset, AddFiles can register Hive-style layouts where a partition value exists only in the folder name:If the table declares
dtrequired and is partitioned bydtas is (identity), schema evolution treateddtas a required column missing from the file, so every row would read null. That meant:ALLOW_FIELD_ADDITIONonly, the pipeline failed (FAIL_PIPELINE) or every file went to the error output (ROUTE_TO_ERRORS);ALLOW_FIELD_RELAXATION,dtwas made optional;dtinrequired_columns, the pipeline failed or every file was refused, even withALLOW_FIELD_RELAXATION.These files don't read null for
dt. Per the Iceberg spec (column projection), a column missing from a data file is read from the file's partition value when an identity partition exists for it. AddFiles sets that value from the folder name.Change
When
location_prefixis set and the table has an identity partition on a column, a file whose path gives that column a value (adt=...folder other thandt=__HIVE_DEFAULT_PARTITION__) no longer counts as missing it:ReadFooterSchemarecords the partition keys in each file's path,CollectDistinctSchemaskeeps the keys every file of a schema has, andSchemaPlanpasses the matching column ids toSchemaDelta.required_columnscheck do the same from the file's own path. The exemption only applies when the file lacks the column; a file that hasdtwith nulls in it is still refused.SchemaDelta.classifytakes a set of columns filled from outside the file and doesn't report them as needing to become optional.PartitionPathshelper; the existing prefix cut moved into it unchanged.dt, files underdt=.../without adtcolumnALLOW_FIELD_ADDITIONonlydtstays requiredALLOW_FIELD_RELAXATIONdtmade optionaldtstays requireddtinrequired_columnsdtstays requireddtvalue)dtmade optional with relaxationROUTE_TO_ERRORSthe files indt=folders now registerUnchanged:
location_prefix, or an unpartitioned table.