Default the Hudi target to table version 6 and parameterize tests for versions 6 and 9

Flip HudiTargetConfig.DEFAULT_TABLE_VERSION from NINE to SIX so the
default output stays readable by released Hudi readers; version 9
remains fully supported via xtable.hudi.target.table_version=9.

Parameterize the Hudi test suites so every run exercises both table
versions instead of only the default:
- ITHudiConversionTarget: partitioned x {SIX, NINE} via a MethodSource
  cross-product; the target client sets the version through
  HudiTargetConfig.HUDI_TABLE_VERSION.
- ITHudiConversionSource: source tables are created at {SIX, NINE} via
  the table-type/partition MethodSource cross-products, and the
  parameterized tests write through TestSparkHudiTable (Spark writer)
  instead of the Java client.
- ITConversionController: combinations targeting HUDI are emitted once
  per version; getTableSyncConfig gained an overload that applies the
  version to the Hudi target properties.
- TestHudiTargetConfig/TestHudiConversionTarget assert against
  DEFAULT_TABLE_VERSION instead of a hard-coded version.

Version 9 source coverage in ITHudiConversionSource depends on two
Hudi fixes validated against a locally patched 1.3.0-SNAPSHOT:
apache/hudi#19126 (column stats on map/array-nested leaves during MOR
log-append) and the savepoint backlog fix in the previous commit.

Author: vinishjail97

xtable-core/src/main/java/org/apache/xtable/hudi/HudiTargetConfig.java

@@ -34,7 +34,7 @@ public class HudiTargetConfig {
    */
   public static final String HUDI_TABLE_VERSION = "xtable.hudi.target.table_version";
 
-  static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.NINE;
+  static final HoodieTableVersion DEFAULT_TABLE_VERSION = HoodieTableVersion.SIX;
 
   HoodieTableVersion tableVersion;
 

xtable-core/src/test/java/org/apache/xtable/ITConversionController.java

@@ -80,6 +80,7 @@
 import org.apache.hudi.common.model.HoodieAvroPayload;
 import org.apache.hudi.common.model.HoodieRecord;
 import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableVersion;
 import org.apache.hudi.common.table.timeline.HoodieInstant;
 
 import org.apache.iceberg.Snapshot;
@@ -101,6 +102,7 @@
 import org.apache.xtable.conversion.TargetTable;
 import org.apache.xtable.delta.DeltaConversionSourceProvider;
 import org.apache.xtable.hudi.HudiConversionSourceProvider;
+import org.apache.xtable.hudi.HudiTargetConfig;
 import org.apache.xtable.hudi.HudiTestUtil;
 import org.apache.xtable.iceberg.IcebergConversionSourceProvider;
 import org.apache.xtable.iceberg.TestIcebergDataHelper;
@@ -153,7 +155,16 @@ private static Stream<Arguments> generateTestParametersForFormatsSyncModesAndPar
     for (String sourceFormat : Arrays.asList(HUDI, DELTA, ICEBERG, PAIMON)) {
       for (SyncMode syncMode : SyncMode.values()) {
         for (boolean isPartitioned : new boolean[] {true, false}) {
-          arguments.add(Arguments.of(sourceFormat, syncMode, isPartitioned));
+          if (sourceFormat.equals(HUDI)) {
+            // Hudi is the source here (not a target), so the Hudi target version does not apply.
+            arguments.add(Arguments.of(sourceFormat, syncMode, isPartitioned, null));
+          } else {
+            // Hudi is one of the targets; exercise both supported target versions.
+            arguments.add(
+                Arguments.of(sourceFormat, syncMode, isPartitioned, HoodieTableVersion.SIX));
+            arguments.add(
+                Arguments.of(sourceFormat, syncMode, isPartitioned, HoodieTableVersion.NINE));
+          }
         }
       }
     }
@@ -225,7 +236,10 @@ private ConversionSourceProvider<?> getConversionSourceProvider(String sourceTab
   @ParameterizedTest
   @MethodSource("generateTestParametersForFormatsSyncModesAndPartitioning")
   public void testVariousOperations(
-      String sourceTableFormat, SyncMode syncMode, boolean isPartitioned) {
+      String sourceTableFormat,
+      SyncMode syncMode,
+      boolean isPartitioned,
+      HoodieTableVersion hudiTargetVersion) {
     String tableName = getTableName();
     List<String> targetTableFormats = getOtherFormats(sourceTableFormat);
     String partitionConfig = null;
@@ -248,7 +262,8 @@ public void testVariousOperations(
               table,
               targetTableFormats,
               partitionConfig,
-              null);
+              null,
+              hudiTargetVersion);
       conversionController.sync(conversionConfig, conversionSourceProvider);
       checkDatasetEquivalence(sourceTableFormat, table, targetTableFormats, 100);
 
@@ -280,7 +295,8 @@ public void testVariousOperations(
               tableWithUpdatedSchema,
               targetTableFormats,
               partitionConfig,
-              null);
+              null,
+              hudiTargetVersion);
       List<Row> insertsAfterSchemaUpdate = tableWithUpdatedSchema.insertRows(100);
       tableWithUpdatedSchema.reload();
       conversionController.sync(conversionConfig, conversionSourceProvider);
@@ -524,43 +540,53 @@ private static Stream<Arguments> provideArgsForPartitionTesting() {
     String severityFilter = "severity = 1";
     String timestampAndLevelFilter = String.format("%s and %s", timestampFilter, levelFilter);
     return Stream.of(
-        Arguments.of(
             buildArgsForPartition(
-                HUDI, Arrays.asList(ICEBERG, DELTA), "level:SIMPLE", "level:VALUE", levelFilter)),
-        Arguments.of(
+                HUDI, Arrays.asList(ICEBERG, DELTA), "level:SIMPLE", "level:VALUE", levelFilter),
             buildArgsForPartition(
-                DELTA, Arrays.asList(ICEBERG, HUDI), null, "level:VALUE", levelFilter)),
-        Arguments.of(
+                DELTA, Arrays.asList(ICEBERG, HUDI), null, "level:VALUE", levelFilter),
             buildArgsForPartition(
-                ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter)),
-        // Delta is excluded here since it does not support nested partition columns.
-        Arguments.of(
+                ICEBERG, Arrays.asList(DELTA, HUDI), null, "level:VALUE", levelFilter),
+            // Delta is excluded here since it does not support nested partition columns.
             buildArgsForPartition(
                 HUDI,
                 Arrays.asList(ICEBERG),
                 "nested_record.level:SIMPLE",
                 "nested_record.level:VALUE",
-                nestedLevelFilter)),
-        Arguments.of(
+                nestedLevelFilter),
             buildArgsForPartition(
                 HUDI,
                 Arrays.asList(ICEBERG, DELTA),
                 "severity:SIMPLE",
                 "severity:VALUE",
-                severityFilter)),
-        Arguments.of(
+                severityFilter),
             buildArgsForPartition(
                 HUDI,
                 Arrays.asList(ICEBERG, DELTA),
                 "timestamp_micros_nullable_field:TIMESTAMP,level:SIMPLE",
                 "timestamp_micros_nullable_field:DAY:yyyy/MM/dd,level:VALUE",
                 timestampAndLevelFilter,
-                getAdditionalHudiReadOptions())));
+                getAdditionalHudiReadOptions()))
+        .flatMap(ITConversionController::withHudiTargetVersions);
+  }
+
+  /**
+   * Expands a partition-test case across both Hudi target versions (6 and 9) when Hudi is one of
+   * the target formats; otherwise yields the single case with no version override.
+   */
+  private static Stream<Arguments> withHudiTargetVersions(TableFormatPartitionDataHolder holder) {
+    if (holder.getTargetTableFormats().contains(HUDI)) {
+      return Stream.of(
+          Arguments.of(holder, HoodieTableVersion.SIX),
+          Arguments.of(holder, HoodieTableVersion.NINE));
+    }
+    return Stream.of(Arguments.of(holder, (HoodieTableVersion) null));
   }
 
   @ParameterizedTest
   @MethodSource("provideArgsForPartitionTesting")
-  public void testPartitionedData(TableFormatPartitionDataHolder tableFormatPartitionDataHolder) {
+  public void testPartitionedData(
+      TableFormatPartitionDataHolder tableFormatPartitionDataHolder,
+      HoodieTableVersion hudiTargetVersion) {
     String tableName = getTableName();
     String sourceTableFormat = tableFormatPartitionDataHolder.getSourceTableFormat();
     List<String> targetTableFormats = tableFormatPartitionDataHolder.getTargetTableFormats();
@@ -587,7 +613,8 @@ public void testPartitionedData(TableFormatPartitionDataHolder tableFormatPartit
               table,
               targetTableFormats,
               xTablePartitionConfig,
-              null);
+              null,
+              hudiTargetVersion);
       tableToClose.insertRows(100);
       conversionController.sync(conversionConfig, conversionSourceProvider);
       // Do a second sync to force the test to read back the metadata it wrote earlier
@@ -1184,6 +1211,26 @@ private static ConversionConfig getTableSyncConfig(
       List<String> targetTableFormats,
       String partitionConfig,
       Duration metadataRetention) {
+    return getTableSyncConfig(
+        sourceTableFormat,
+        syncMode,
+        tableName,
+        table,
+        targetTableFormats,
+        partitionConfig,
+        metadataRetention,
+        null);
+  }
+
+  private static ConversionConfig getTableSyncConfig(
+      String sourceTableFormat,
+      SyncMode syncMode,
+      String tableName,
+      GenericTable table,
+      List<String> targetTableFormats,
+      String partitionConfig,
+      Duration metadataRetention,
+      HoodieTableVersion hudiTargetVersion) {
     Properties sourceProperties = new Properties();
     if (partitionConfig != null) {
       sourceProperties.put(PARTITION_FIELD_SPEC_CONFIG, partitionConfig);
@@ -1207,7 +1254,7 @@ private static ConversionConfig getTableSyncConfig(
                         // set the metadata path to the data path as the default (required by Hudi)
                         .basePath(table.getDataPath())
                         .metadataRetention(metadataRetention)
-                        .additionalProperties(new TypedProperties())
+                        .additionalProperties(hudiTargetProperties(formatName, hudiTargetVersion))
                         .build())
             .collect(Collectors.toList());
 
@@ -1217,4 +1264,18 @@ private static ConversionConfig getTableSyncConfig(
         .syncMode(syncMode)
         .build();
   }
+
+  /**
+   * Returns the additional properties for a target table, pinning the Hudi target table version
+   * when one is supplied (and the target is Hudi) so a single test can exercise both v6 and v9.
+   */
+  private static TypedProperties hudiTargetProperties(
+      String formatName, HoodieTableVersion hudiTargetVersion) {
+    TypedProperties properties = new TypedProperties();
+    if (HUDI.equals(formatName) && hudiTargetVersion != null) {
+      properties.setProperty(
+          HudiTargetConfig.HUDI_TABLE_VERSION, String.valueOf(hudiTargetVersion.versionCode()));
+    }
+    return properties;
+  }
 }

xtable-core/src/test/java/org/apache/xtable/TestSparkHudiTable.java

@@ -47,6 +47,7 @@
 import org.apache.hudi.common.model.HoodieTableType;
 import org.apache.hudi.common.model.WriteOperationType;
 import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
 import org.apache.hudi.common.util.CommitUtils;
 import org.apache.hudi.common.util.Option;
 import org.apache.hudi.config.HoodieWriteConfig;
@@ -115,6 +116,17 @@ public static TestSparkHudiTable forStandardSchema(
         tableName, BASIC_SCHEMA, tempDir, jsc, partitionConfig, tableType);
   }
 
+  public static TestSparkHudiTable forStandardSchema(
+      String tableName,
+      Path tempDir,
+      JavaSparkContext jsc,
+      String partitionConfig,
+      HoodieTableType tableType,
+      HoodieTableVersion tableVersion) {
+    return new TestSparkHudiTable(
+        tableName, BASIC_SCHEMA, tempDir, jsc, partitionConfig, tableType, tableVersion);
+  }
+
   /**
    * Create a test table instance with a schema that has more fields than an instance returned by
    * {@link #forStandardSchema(String, Path, JavaSparkContext, String, HoodieTableType)}.
@@ -153,7 +165,20 @@ private TestSparkHudiTable(
       JavaSparkContext jsc,
       String partitionConfig,
       HoodieTableType hoodieTableType) {
+    this(name, schema, tempDir, jsc, partitionConfig, hoodieTableType, HoodieTableVersion.SIX);
+  }
+
+  private TestSparkHudiTable(
+      String name,
+      Schema schema,
+      Path tempDir,
+      JavaSparkContext jsc,
+      String partitionConfig,
+      HoodieTableType hoodieTableType,
+      HoodieTableVersion tableVersion) {
     super(name, schema, tempDir, partitionConfig);
+    // set the table version before initializing the write/meta clients, which read it
+    this.tableVersion = tableVersion;
     // initialize spark session
     this.jsc = jsc;
     this.writeClient = initSparkWriteClient(schema, typedProperties);