feat: add native Parquet VARIANT column support for JSON and semi-structured fields

Write Kafka Connect JSON fields (Debezium CDC, Confluent Protobuf Struct,
custom messages, maps, arrays) as native Parquet VARIANT columns instead of
plain STRING, improving storage efficiency and query performance.

- Upgrade parquet-java to 1.17.0 for VARIANT logical type support - https://parquet.apache.org/blog/2026/02/27/variant-type-in-apache-parquet-for-semi-structured-data/
- Add config: parquet.variant.enabled, parquet.variant.connect.names,
  parquet.variant.field.names
- Auto-detect recursive schemas (google.protobuf.Struct) as VARIANT
- Stream JSON-to-Variant conversion ported from Apache Spark (PR also raised in parquet-java, once approved, code will be neat here in this repo: apache/parquet-java#3415)
- Unwrap Protobuf Struct/Value/ListValue/map-as-array to clean JSON
- Graceful fallback for non-JSON values (e.g. __debezium_unavailable_value)
- Feature is fully opt-in (disabled by default), zero impact on existing connectors

Author: gaurav7261

kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java

@@ -238,6 +238,45 @@ public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
   public static final String TOMBSTONE_ENCODED_PARTITION = "tombstone.encoded.partition";
   public static final String TOMBSTONE_ENCODED_PARTITION_DEFAULT = "tombstone";
 
+  // Parquet Variant configs
+  public static final String PARQUET_VARIANT_ENABLED_CONFIG = "parquet.variant.enabled";
+  public static final boolean PARQUET_VARIANT_ENABLED_DEFAULT = false;
+  public static final String PARQUET_VARIANT_ENABLED_DOC =
+      "When enabled, fields identified as carrying semi-structured "
+          + "data will be written as Parquet VARIANT columns. "
+          + "Fields with recursive schemas (e.g. Protobuf "
+          + "google.protobuf.Struct) are auto-detected and "
+          + "converted to VARIANT, preventing StackOverflowError "
+          + "in AvroSchemaConverter. Additionally supports STRING "
+          + "fields with JSON (Debezium io.debezium.data.Json), "
+          + "complex STRUCT/MAP fields, and ARRAY fields. "
+          + "Provides type-preserving binary encoding, efficient "
+          + "field access, and better query performance in "
+          + "downstream engines "
+          + "(Spark, DuckDB, Snowflake, Databricks, Trino).";
+
+  public static final String PARQUET_VARIANT_CONNECT_NAMES_CONFIG =
+      "parquet.variant.connect.names";
+  public static final String PARQUET_VARIANT_CONNECT_NAMES_DEFAULT =
+      "io.debezium.data.Json";
+  public static final String PARQUET_VARIANT_CONNECT_NAMES_DOC =
+      "Comma-separated list of Kafka Connect schema names whose "
+          + "fields should be written as Parquet VARIANT columns. "
+          + "Works for any field type (STRING, STRUCT, MAP, ARRAY).";
+
+  public static final String PARQUET_VARIANT_FIELD_NAMES_CONFIG =
+      "parquet.variant.field.names";
+  public static final String PARQUET_VARIANT_FIELD_NAMES_DEFAULT = "";
+  public static final String PARQUET_VARIANT_FIELD_NAMES_DOC =
+      "Comma-separated list of field names to explicitly write as "
+          + "Parquet VARIANT columns, regardless of their Connect "
+          + "schema name or type. Note: fields with recursive "
+          + "schemas (google.protobuf.Struct) are auto-detected "
+          + "and do NOT need to be listed here. Use this for "
+          + "STRING fields containing JSON (e.g. model inference "
+          + "logs) or custom Protobuf messages that you want "
+          + "stored as VARIANT.";
+
   /**
    * Append schema name in s3-path
    */
@@ -820,6 +859,47 @@ public static ConfigDef newConfigDef() {
       );
     }
 
+    {
+      final String group = "Parquet Variant";
+      int orderInGroup = 0;
+
+      configDef.define(
+          PARQUET_VARIANT_ENABLED_CONFIG,
+          Type.BOOLEAN,
+          PARQUET_VARIANT_ENABLED_DEFAULT,
+          Importance.MEDIUM,
+          PARQUET_VARIANT_ENABLED_DOC,
+          group,
+          ++orderInGroup,
+          Width.SHORT,
+          "Enable Parquet Variant for JSON fields"
+      );
+
+      configDef.define(
+          PARQUET_VARIANT_CONNECT_NAMES_CONFIG,
+          Type.LIST,
+          PARQUET_VARIANT_CONNECT_NAMES_DEFAULT,
+          Importance.LOW,
+          PARQUET_VARIANT_CONNECT_NAMES_DOC,
+          group,
+          ++orderInGroup,
+          Width.LONG,
+          "Connect schema names to treat as Variant"
+      );
+
+      configDef.define(
+          PARQUET_VARIANT_FIELD_NAMES_CONFIG,
+          Type.LIST,
+          PARQUET_VARIANT_FIELD_NAMES_DEFAULT,
+          Importance.LOW,
+          PARQUET_VARIANT_FIELD_NAMES_DOC,
+          group,
+          ++orderInGroup,
+          Width.LONG,
+          "Explicit field names to write as Variant"
+      );
+    }
+
     {
       final String group = "Keys and Headers";
       int orderInGroup = 0;
@@ -1418,6 +1498,28 @@ public boolean shouldRotateOnPartitionChange() {
     return getBoolean(ROTATE_FILE_ON_PARTITION_CHANGE);
   }
 
+  public boolean isParquetVariantEnabled() {
+    return getBoolean(PARQUET_VARIANT_ENABLED_CONFIG);
+  }
+
+  public List<String> getParquetVariantConnectNames() {
+    return getList(PARQUET_VARIANT_CONNECT_NAMES_CONFIG);
+  }
+
+  public Set<String> getParquetVariantFieldNames() {
+    List<String> list = getList(PARQUET_VARIANT_FIELD_NAMES_CONFIG);
+    if (list == null) {
+      return Collections.emptySet();
+    }
+    Set<String> result = new HashSet<>();
+    for (String s : list) {
+      if (s != null && !s.trim().isEmpty()) {
+        result.add(s.trim());
+      }
+    }
+    return result;
+  }
+
   public enum IgnoreOrFailBehavior {
     IGNORE,
     FAIL;

kafka-connect-s3/src/main/java/io/confluent/connect/s3/format/parquet/ParquetRecordWriterProvider.java

@@ -25,6 +25,8 @@
 import io.confluent.connect.s3.storage.IORecordWriter;
 import io.confluent.connect.s3.format.RecordViewSetter;
 import io.confluent.connect.s3.format.S3RetriableRecordWriter;
+import io.confluent.connect.s3.format.parquet.variant.JsonFieldDetector;
+import io.confluent.connect.s3.format.parquet.variant.VariantAwareWriteSupport;
 import io.confluent.connect.s3.storage.S3ParquetOutputStream;
 import io.confluent.connect.s3.storage.S3Storage;
 import io.confluent.connect.storage.format.RecordWriter;
@@ -37,12 +39,14 @@
 import org.apache.parquet.avro.AvroWriteSupport;
 import org.apache.parquet.hadoop.ParquetFileWriter;
 import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
 import org.apache.parquet.io.OutputFile;
 import org.apache.parquet.io.PositionOutputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -66,12 +70,15 @@ public String getExtension() {
 
   @Override
   public RecordWriter getRecordWriter(final S3SinkConnectorConfig conf, final String filename) {
+    final boolean variantEnabled = conf.isParquetVariantEnabled();
+
     return new S3RetriableRecordWriter(
         new IORecordWriter() {
           final String adjustedFilename = getAdjustedFilename(recordView, filename, getExtension());
           Schema schema = null;
           ParquetWriter<GenericRecord> writer;
           S3ParquetOutputFile s3ParquetOutputFile;
+          Set<String> variantFieldPaths = Collections.emptySet();
 
           @Override
           public void write(SinkRecord record) throws IOException {
@@ -80,26 +87,49 @@ public void write(SinkRecord record) throws IOException {
               log.info("Opening record writer for: {}", adjustedFilename);
               org.apache.avro.Schema avroSchema = avroData.fromConnectSchema(schema);
               s3ParquetOutputFile = new S3ParquetOutputFile(storage, adjustedFilename);
-              AvroParquetWriter.Builder<GenericRecord> builder =
-                  AvroParquetWriter.<GenericRecord>builder(s3ParquetOutputFile)
-                      .withSchema(avroSchema)
-                      .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
-                      .withDictionaryEncoding(true)
-                      .withCompressionCodec(storage.conf().parquetCompressionCodecName())
-                      .withPageSize(PAGE_SIZE);
-              if (schemaHasArrayOfOptionalItems(schema, /*seenSchemas=*/null)) {
-                // If the schema contains an array of optional items, then
-                // it is possible that the array may have null items during the
-                // writing process.  In this case, we set a flag so as not to
-                // incur a NullPointerException
+
+              boolean useOldListStructure = !schemaHasArrayOfOptionalItems(
+                  schema, /*seenSchemas=*/null
+              );
+              if (!useOldListStructure) {
                 log.debug(
                     "Setting \"" + AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE
                         + "\" to false because the schema contains an array "
                         + "with optional items"
                 );
-                builder.config(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false");
               }
-              writer = builder.build();
+
+              if (variantEnabled) {
+                variantFieldPaths = detectVariantFields(conf, schema);
+              }
+
+              if (variantEnabled && !variantFieldPaths.isEmpty()) {
+                log.info("Variant-aware Parquet writer enabled for fields: {}",
+                    variantFieldPaths);
+                VariantAwareWriteSupport writeSupport = new VariantAwareWriteSupport(
+                    avroSchema,
+                    variantFieldPaths,
+                    useOldListStructure
+                );
+                writer = new VariantParquetWriterBuilder(s3ParquetOutputFile, writeSupport)
+                    .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+                    .withDictionaryEncoding(true)
+                    .withCompressionCodec(storage.conf().parquetCompressionCodecName())
+                    .withPageSize(PAGE_SIZE)
+                    .build();
+              } else {
+                AvroParquetWriter.Builder<GenericRecord> builder =
+                    AvroParquetWriter.<GenericRecord>builder(s3ParquetOutputFile)
+                        .withSchema(avroSchema)
+                        .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+                        .withDictionaryEncoding(true)
+                        .withCompressionCodec(storage.conf().parquetCompressionCodecName())
+                        .withPageSize(PAGE_SIZE);
+                if (!useOldListStructure) {
+                  builder.config(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false");
+                }
+                writer = builder.build();
+              }
             }
             log.trace("Sink record with view {}: {}", recordView,
                 sinkRecordToLoggableString(record));
@@ -125,6 +155,14 @@ public void commit() throws IOException {
     );
   }
 
+  private static Set<String> detectVariantFields(S3SinkConnectorConfig conf, Schema schema) {
+    JsonFieldDetector detector = new JsonFieldDetector(
+        conf.getParquetVariantConnectNames(),
+        conf.getParquetVariantFieldNames()
+    );
+    return detector.detect(schema);
+  }
+
   /**
    * Check if any schema (or nested schema) is an array of optional items
    * @param schema The shema to check
@@ -157,6 +195,36 @@ public static boolean schemaHasArrayOfOptionalItems(Schema schema, Set<Schema> s
     }
   }
 
+  private static class VariantParquetWriterBuilder
+      extends ParquetWriter.Builder<GenericRecord, VariantParquetWriterBuilder> {
+
+    private final VariantAwareWriteSupport writeSupport;
+
+    VariantParquetWriterBuilder(OutputFile outputFile, VariantAwareWriteSupport writeSupport) {
+      super(outputFile);
+      this.writeSupport = writeSupport;
+    }
+
+    @Override
+    protected VariantParquetWriterBuilder self() {
+      return this;
+    }
+
+    @Override
+    protected WriteSupport<GenericRecord> getWriteSupport(
+        org.apache.hadoop.conf.Configuration conf
+    ) {
+      return writeSupport;
+    }
+
+    @Override
+    protected WriteSupport<GenericRecord> getWriteSupport(
+        org.apache.parquet.conf.ParquetConfiguration conf
+    ) {
+      return writeSupport;
+    }
+  }
+
   private static class S3ParquetOutputFile implements OutputFile {
     private static final int DEFAULT_BLOCK_SIZE = 0;
     private S3Storage storage;