fix: Iceberg commit message when writer rotated

[PingLiuPing]

Fixes commit message correctness for multi-file writes by updating Iceberg writer rotation handling and tracking per-file row counts and emitting one commit task per rotated file.
Fix parquet writer does not honor smaller values of max-target-file-size config.

These are the adjustments required after #16077.

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	b8956f6
🔍 Latest deploy log	https://app.netlify.com/projects/meta-velox/deploys/698e518d6e97a200070e67b4

[mbasmanova]

@PingLiuPing Is it possible to add a test to protect this behavior?

[mbasmanova]

nit: rowCount -> numRows

[PingLiuPing]

Thanks.

[mbasmanova]

// clang-format off

this should go inside the loop

[PingLiuPing]

Thanks.

[PingLiuPing]

@PingLiuPing Is it possible to add a test to protect this behavior?

@mbasmanova Thank you for the quick review.

I attempted to add a test for this behavior. And found that triggering writer rotation requires writing a large amount of data, since the default Parquet flush policy rotates only after 1,024,000 rows or 128MB.

One possible approach is to allow IcebergInsertTableHandle to accept WriterOptions, so that tests can inject customized writer options with a smaller flush threshold. For example:

auto writerOptions = std::make_shared<parquet::WriterOptions>();
writerOptions->compressionKind = common::CompressionKind::CompressionKind_NONE;
writerOptions->enableDictionary = false;
writerOptions->flushPolicyFactory = []() {
  return std::make_unique<parquet::DefaultFlushPolicy>(256, 8 * 1024);
};

This would allow us to trigger writer rotation with a small dataset in tests.

Let me explore whether there is a cleaner or more appropriate approach before proceeding.

[PingLiuPing]

@mbasmanova

Ideally, we should be able to set kMaxTargetFileSizeSession to an arbitrary value and rely on the writer to rotate once the file size reaches that limit.
But, rotation depends on ioStats_->rawBytesWritten(), which only advances when the writer flushes.
I think this configuration does not currently work correctly for smaller values (e.g., values smaller than the default Parquet flush threshold bytesInRowGroup_). In such cases, setting kMaxTargetFileSizeSession to a smaller value does not result in generating files close to the expected size.

This is likely because Parquet’s flush logic is driven by row group thresholds (row count or row group size), rather than the target file size:

bool shouldFlush(
    const dwio::common::StripeProgress& stripeProgress) override {
  return stripeProgress.stripeRowCount >= rowsInRowGroup_ ||
         stripeProgress.stripeSizeEstimate >= bytesInRowGroup_;
}

This issue may also exist in the DWRF writer, although its flush policy is more complex. Parquet’s behavior is relatively straightforward since it is purely based on row count and row group size.

Would it make sense for me to open a separate PR to address this issue and add proper test coverage there?

[mbasmanova]

Would it make sense for me to open a separate PR to address this issue and add proper test coverage there?

Thank would be nice. Thanks.

[meta-codesync]

@jainxrohit has imported this pull request. If you are a Meta employee, you can view this in D93157234.

[meta-codesync]

@jainxrohit merged this pull request in 9cb3cd1.