This is an automated email from the ASF dual-hosted git repository.
vinishjail97 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git
The following commit(s) were added to refs/heads/main by this push:
new d34a37a0 fix(spark-runtime): Support Parquet source partitioning
(#927) (#931)
d34a37a0 is described below
commit d34a37a0eb4bacb41ef37bb699a37704cfae2d7d
Author: Arvin <[email protected]>
AuthorDate: Fri Oct 23 13:18:20 2026 +0800
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
---
.../main/java/org/apache/xtable/delta/DeltaConversionSource.java | 1 +
.../org/apache/xtable/kernel/DeltaKernelConversionSource.java | 1 +
.../src/main/java/org/apache/xtable/spark/TableSyncSpec.java | 6 +++++-
.../src/main/java/org/apache/xtable/spark/XTableSparkSync.java | 3 ++-
.../src/main/java/org/apache/xtable/spark/XTableSyncService.java | 8 +++++++-
5 files changed, 16 insertions(+), 3 deletions(-)
diff --git
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
b/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
index fc1d17c6..2a6f8958 100644
---
a/xtable-core/src/main/java/org/apache/xtable/delta/DeltaConversionSource.java
+++
b/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;
diff --git
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
b/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
index 14241f92..ca0f4b94 100644
---
a/xtable-core/src/main/java/org/apache/xtable/kernel/DeltaKernelConversionSource.java
+++
b/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();
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
index 75d35a32..4aeb4daa 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/TableSyncSpec.java
+++
b/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}). */
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
index 84a8b863..cf2f7877 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSparkSync.java
+++
b/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()
diff --git
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
index 43eb720a..e875dab7 100644
---
a/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
+++
b/xtable-spark-runtime/src/main/java/org/apache/xtable/spark/XTableSyncService.java
@@ -44,6 +44,7 @@ import org.apache.xtable.model.sync.SyncMode;
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()
