Data, Kafka Connect: Enable Parquet variant shredding for Record writes

[nssalian]

Closes #16387

Rationale for the change

Kafka Connect and other generic Record writers can't shred variant columns GenericFormatModels registers the Record Parquet model with no analyzer, so write.parquet.shred-variants=true silently does nothing.

Changes

• New RecordVariantShreddingAnalyzer: stateless, resolves variant columns by position
  in Schema.columns().
• GenericFormatModels registers it for the Record Parquet model.
• ParquetFormatModel: opt-in shouldDeriveEngineSchema() hook derives the engine schema from the Iceberg schema when none is set (order-independent).
• RegistryBasedFileWriterFactory: sets engineSchema only when input schema is non-null.
• Docs: note per-file shred layout and the tunable inference buffer size.

Note this continues forward @soumilshah1995's #16370 whilst incorporating the feedback on it. Thanks Soumil for the initial work on this.

Testing

• New TestRecordVariantShreddingAnalyzer: tests for column resolution, null/non-Variant handling, Kafka connect round-trip via GenericFileWriterFactory, and the buffer-flush boundary.
• TestRecordConverter: verifies the converter emits records in table-schema field order regardless of source field order.

[Guosmilesmile]

Thanks for the PR.

Generic, Flink, and Spark all seem to be concrete, engine-specific implementations. Could we get ride of deriveEngineSchema flag to the shared ParquetFormatModel ？

Would it make sense to keep this fallback in the corresponding engine-specific implementation instead, and only fall back there when engineSchema is null?

Also, would it make sense to align this with the approach in #15688?

[nssalian]

Thank for taking a look @Guosmilesmile. Appreciate the pointer to that PR. I dropped the deriveEngineSchema flag so the shared ParquetFormatModel and factory are back to plain setters. The fallback now lives in RecordVariantShreddingAnalyzer and only fires when engineSchema is null, falling back to the Iceberg schema (same thing for Record), which aligns with the engine-specific placement in the PR you mentioned. I put it in the analyzer rather than the factory since that's the one point both the FormatModel and registry-direct write paths share, so neither can hit a null at resolveColumnIndex.

[Guosmilesmile]

LGTM overall. I left a few small comments, but no major concerns.

[laskoviymishka]

The layering reads much better now that the fallback lives in RecordVariantShreddingAnalyzer.analyzeVariantColumns rather than a boolean on the shared ParquetFormatModel — that's the shape @Guosmilesmile was after, and the nice part is the write path stays untouched. Thanks for reworking it.

The tests are also doing the thing I usually end up asking for: assertAllRawParquetRowsShredded and the typed_value schema assertions check the physical layout rather than round-tripping and calling that proof, and the explicit-engine-schema pair has a real positive control. That's a lot of coverage for a small analyzer.

What's left is small and I don't think any of it blocks:

• the instanceof Variant guard in extractVariantValues hard-fails the write where every other branch on this path degrades — for the Kafka Connect case I'd sooner skip shredding that column than fail the commit
• the residual test proves the schema dropped c but not that row 3's c actually landed in the residual value — a raw-level assertion would close it
• the round-trip test exercises GenericFileWriterFactory with hand-built records rather than RecordConverter, which is the one path the rationale is actually about
• the explicit-engine-schema tests only cover the rename-miss case; a reordered engine schema isn't exercised, and that's the one where positions can diverge

Details inline. None of it is layering — happy to see this land once these are sorted.

[laskoviymishka]

Everything else on this path degrades: a missing column warns and skips, an all-null buffer falls back to unshredded. This one throws, so a single unexpected object at a variant position fails the whole write rather than writing it unshredded. For the Kafka Connect case, where the Record is assembled from whatever the converter produced, I'd rather log and skip shredding for that column than fail the commit. Spark's analyzer doesn't check here at all and Flink throws UnsupportedOperationException, so there's no established behavior to match — but a hard failure from the most permissive of the three writers is the combination I'd least expect.

[nssalian]

Kept it as a hard fail on purpose. On the Kafka Connect path the converter always produces a variant, so this only trips on a hand-built bad record, and Flink throws here too. Skipping wouldn't save the write anyway - the writer still needs a variant, so it'd just fail later with a murkier error. Let me know what you think.