lintingbin commented on PR #4188: URL: https://github.com/apache/amoro/pull/4188#issuecomment-4277591369
Hi @zhangwl9, thanks for the PR! I reviewed it against the ongoing Iceberg community work on the same feature ([apache/iceberg#14435](https://github.com/apache/iceberg/pull/14435)) to cross-check which concerns apply here. Sharing my findings below. ## Context: why not just wait for Iceberg? Iceberg #14435 is still in flight — row-lineage handling (V3 `_row_id` / `_last_updated_sequence_number`), API shape, and the `expectedOutputFiles==1` question are all unresolved as of the latest round of comments. Even once it lands, its entry point is `SparkParquetFileMergeRunner` (driver/executor split, `TaskContext.taskAttemptId()`), which Amoro's engine-agnostic optimizer cannot reuse directly. @pvary suggested moving the checks/merge core into a common `ParquetFileMerger` in iceberg-core so Flink and others can share it, but that refactor hasn't landed yet either. So I think shipping an Amoro-native implementation now is the right call, **as a transitional impl** that we can replace once Iceberg exposes a reusable core. That framing changes what "must fix before merge" means: ## Recommendation: scope this PR to V2 tables and keep the surface minimal If we explicitly skip V3 row-lineage tables (fallback gate), the big Iceberg-side unknowns (`_row_id` preservation, `_last_updated_sequence_number` inheritance, null-stats edge cases) simply don't apply to us, and we can ship a much tighter patch. ## Must-fix before merge (4 items) 1. **Add a V3 row-lineage gate.** If the table is V3 and row lineage is enabled, return `false` from `canParquetRowGroupMerge()` so we fall back. This keeps us out of territory the Iceberg PR hasn't finalized. 2. **`parquetRowGroupMergeFiles()` should call `this.targetSize()`, not `super.targetSize()`.** `IcebergRewriteExecutor.targetSize()` already overrides to return `Long.MAX_VALUE` when the input total is below target (see [#3645](https://github.com/apache/amoro/issues/3645)). Bypassing the override makes the row-group-merge path split into multiple files where the row-based path would consolidate into one — inconsistent with #3645's intent. 3. **Clean up already-written output files when fallback is triggered.** Once a rolled-over `ParquetFileMergeRunner.result()` has succeeded, the corresponding `DataFile` is appended to `outputFiles`; if a later runner throws and we `catch → super.rewriterDataFiles()`, those earlier physical files are never deleted. `ParquetFileMergeRunner.close()` only handles the *current* runner. Please add a best-effort `io.deleteFile()` sweep over the accumulated `outputFiles` in the catch/finally. 4. **`hasNoBloomFilter()` misses the global switch.** It only inspects `write.parquet.bloom-filter-enabled.column.*` (column prefix). Iceberg also supports the table-wide `write.parquet.bloom-filter-enabled.default` — tables with that on would silently lose bloom filters after merge. Please check both. ## Deferrable (track against Iceberg's final decisions) These are real issues but either safe-by-design or minor, and fixing them now risks diverging from whatever Iceberg finalizes. I'd defer: - **Schema comparison via `MessageType.equals()`** — Iceberg comments argue for field-id-based comparison. `MessageType.equals()` is *stricter* (more false negatives → more fallbacks), so it errs on the safe side. Align with Iceberg once the upstream shape stabilizes. - **Deprecated 7-arg `ParquetFileWriter` constructor** — same issue Guosmilesmile flagged on the Iceberg PR; they ended up wrapping Iceberg's own `ParquetWriter`. Works today; worth replacing when we pull their approach. - **`columnIndexTruncateLength` passed as `null`** — Parquet falls back to an internal default. Safe, but should use `ParquetProperties.DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH` once we touch this area again. - **Each input file is opened twice** (once in `checkSchemaAndBuildContext`, once by `ParquetFileWriter.appendFile`) — perf concern on object stores, not correctness. Iceberg's final form propagates the parsed metadata to avoid re-reads. - **`rowGroupSize` param is a no-op for `appendFile()`** — byte-level row-group copies preserve the source file's row-group size. Please add a note in `configurations.md` so users don't expect this property to resize groups after merge. - **Schema check happens inside `parquetRowGroupMergeFiles()` and throws `IllegalStateException`** (see `testParquetRowGroupMergeWithSchemaDifferent`) — functionally fine today because no writer is open yet when the check returns null, but it's fragile. A small refactor to move schema validation into the gate would eliminate the try/catch-and-retry shape entirely. - **No per-input-file size guard.** The gate doesn't reject inputs that are already ≥ target size, so we still byte-copy them via `appendFile`. On the Iceberg PR, pvary also pointed out that row-group merge shouldn't be used to split big files into smaller ones. Low impact (copy is cheap) but a cheap gate (e.g. skip if a single input already exceeds `targetSize * 0.9`) would avoid unnecessary work and align with the Iceberg discussion. ## Nits (optional) - `checkCondition` logs skip reasons at DEBUG — consider one aggregated INFO line per task so operators can see why the merge path was declined. - `ParquetIOBridge` could get a `// TODO: remove once Iceberg exposes ParquetIO.file() as public` pointing at #14435. - `AbstractRewriteFilesExecutor.rewriterDataFiles()` becoming `protected` extends the subclass API — consider a named hook like `rewriteDataFilesRowBased()` instead. - `MetricsConfig.forTable(table.asUnkeyedTable())` — worth double-checking behavior if the parent `MixedTable` is keyed; `encryptionManager()` in the base class has a keyed/unkeyed split for this reason. ## Positive notes - Gate is conservative in the right way — encryption, deletes, sort order, spec evolution, format, bloom filter (modulo #4 above) are all caught. - Fallback wrapping (`try/catch Exception → super.rewriterDataFiles()`) gives a safety net even if a gate misses something. - Test coverage mirrors the Iceberg discussion well (encryption, bloom filter, sort order, spec evolution, schema mismatch, end-to-end split-and-commit data integrity). Thanks again — the direction is solid, and with the four must-fix items plus a V3 gate I think this is a good transitional implementation that we can later swap for Iceberg's core `ParquetFileMerger` once it lands. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
