fix(spark-runtime): Support Parquet source partitioning (#927) (#931)

* [Bug] Route partition spec to source-format-specific config key (#927)

* fix: add @Builder.Default for initialized fields in conversion sources

* fix: apply spotless formatting to XTableSparkSync.java

Author: zhang-arvin

xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java

@@ -73,6 +73,7 @@ public class DeltaConversionSource implements ConversionSource<Long> {
   @Builder.Default
   private final DeltaTableExtractor tableExtractor = DeltaTableExtractor.builder().build();
 
+  @Builder.Default
   private Optional<DeltaIncrementalChangesState> deltaIncrementalChangesState = Optional.empty();
 
   private final SparkSession sparkSession;

xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java

@@ -74,6 +74,7 @@ public class DeltaKernelConversionSource implements ConversionSource<Long> {
   private final DeltaKernelTableExtractor tableExtractor =
       DeltaKernelTableExtractor.builder().build();
 
+  @Builder.Default
   private Optional<DeltaKernelIncrementalChangesState> deltaKernelIncrementalChangesState =
       Optional.empty();
 

xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java

@@ -40,7 +40,11 @@ public class TableSyncSpec {
   /** Optional namespace segments for the table. */
   String[] namespace;
 
-  /** Optional partition spec for a partitioned Hudi source, e.g. {@code level:VALUE}. */
+  /**
+   * Optional source partition spec (e.g. {@code level:VALUE}); applies to partitioned Hudi and
+   * Parquet sources. Maps to {@code xtable.hudi.source.partition_field_spec_config} for Hudi and
+   * {@code xtable.parquet.source.partition_field_spec_config} for Parquet.
+   */
   String partitionSpec;
 
   /** The source table format, e.g. {@code HUDI} (see {@code TableFormat}). */

xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java

@@ -153,7 +153,8 @@ public final class XTableSparkSync {
               Option.builder()
                   .longOpt(PARTITION_SPEC)
                   .hasArg()
-                  .desc("The Hudi source partition field spec, e.g. level:VALUE")
+                  .desc(
+                      "The source partition field spec, e.g. level:VALUE (Hudi and Parquet sources)")
                   .build())
           .addOption(
               Option.builder()

xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java

@@ -44,6 +44,7 @@
 import org.apache.xtable.model.sync.SyncResult;
 import org.apache.xtable.paimon.PaimonConversionSourceProvider;
 import org.apache.xtable.parquet.ParquetConversionSourceProvider;
+import org.apache.xtable.parquet.ParquetSourceConfig;
 
 /**
  * Runs an incremental {@link ConversionController#sync} for a {@link TableSyncSpec}. The sync falls
@@ -57,7 +58,12 @@ public class XTableSyncService {
   public Map<String, SyncResult> sync(TableSyncSpec spec, Configuration hadoopConf) {
     Properties sourceProperties = new Properties();
     if (spec.getPartitionSpec() != null && !spec.getPartitionSpec().isEmpty()) {
-      sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+      if (TableFormat.PARQUET.equals(spec.getSourceFormat())) {
+        sourceProperties.put(
+            ParquetSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+      } else {
+        sourceProperties.put(HudiSourceConfig.PARTITION_FIELD_SPEC_CONFIG, spec.getPartitionSpec());
+      }
     }
     String dataPath =
         spec.getDataPath() != null && !spec.getDataPath().isEmpty()