This is an automated email from the ASF dual-hosted git repository.

timbrown pushed a commit to branch 412-metadata
in repository https://gitbox.apache.org/repos/asf/incubator-xtable.git

commit e2289dde9db0f6bd97d93c4809a16b04f27b7a06
Author: Timothy Brown <[email protected]>
AuthorDate: Wed Apr 24 15:09:20 2024 -0500

    Move metadata to single json field
---
 api/pom.xml                                        |  14 ++-
 .../org/apache/xtable/model/TableSyncMetadata.java | 104 ---------------------
 .../xtable/model/metadata/TableSyncMetadata.java   |  93 ++++++++++++++++++
 .../apache/xtable/spi/sync/ConversionTarget.java   |   2 +-
 .../apache/xtable/spi/sync/TableFormatSync.java    |   2 +-
 .../model/metadata/TestTableSyncMetadata.java      |  79 ++++++++++++++++
 .../xtable/spi/sync/TestTableFormatSync.java       |   2 +-
 .../xtable/conversion/ConversionController.java    |   2 +-
 .../apache/xtable/delta/DeltaConversionTarget.java |  14 ++-
 .../apache/xtable/hudi/HudiConversionTarget.java   |  11 ++-
 .../xtable/iceberg/IcebergConversionTarget.java    |  18 +---
 .../conversion/TestConversionController.java       |   2 +-
 .../xtable/hudi/ITHudiConversionSourceTarget.java  |   2 +-
 .../hudi/TestHudiConversionSourceTarget.java       |   2 +-
 .../org/apache/xtable/iceberg/TestIcebergSync.java |   2 +-
 15 files changed, 210 insertions(+), 139 deletions(-)

diff --git a/api/pom.xml b/api/pom.xml
index 21d1af61..2d591465 100644
--- a/api/pom.xml
+++ b/api/pom.xml
@@ -50,6 +50,16 @@
             <artifactId>log4j-1.2-api</artifactId>
         </dependency>
 
+        <!--Jackson -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.datatype</groupId>
+            <artifactId>jackson-datatype-jsr310</artifactId>
+        </dependency>
+
         <!-- Junit -->
         <dependency>
             <groupId>org.junit.jupiter</groupId>
@@ -70,9 +80,5 @@
             <artifactId>mockito-core</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-databind</artifactId>
-        </dependency>
     </dependencies>
 </project>
diff --git a/api/src/main/java/org/apache/xtable/model/TableSyncMetadata.java 
b/api/src/main/java/org/apache/xtable/model/TableSyncMetadata.java
deleted file mode 100644
index 7c0b8c1a..00000000
--- a/api/src/main/java/org/apache/xtable/model/TableSyncMetadata.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
- 
-package org.apache.xtable.model;
-
-import java.time.Instant;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.stream.Collectors;
-
-import lombok.AllArgsConstructor;
-import lombok.Value;
-
-/**
- * Metadata representing the state of a table sync process. This metadata is 
stored in the target
- * table's properties and is used to track the status of previous sync 
operation.
- */
-@AllArgsConstructor(staticName = "of")
-@Value
-public class TableSyncMetadata {
-  /**
-   * Property name for the lastInstantSynced field from SyncResult, used for 
persisting
-   * lastInstantSynced in the table metadata/properties
-   */
-  public static final String XTABLE_LAST_INSTANT_SYNCED_PROP = 
"XTABLE_LAST_INSTANT_SYNCED";
-  /**
-   * Property name for the list of instants to consider during the next sync. 
This list may include
-   * out-of-order instants that could be missed without explicit tracking.
-   */
-  public static final String INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP =
-      "INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC";
-
-  Instant lastInstantSynced;
-  List<Instant> instantsToConsiderForNextSync;
-
-  public Map<String, String> asMap() {
-    Map<String, String> map = new HashMap<>();
-    map.put(XTABLE_LAST_INSTANT_SYNCED_PROP, lastInstantSynced.toString());
-    map.put(
-        INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP,
-        convertInstantsToConsiderForNextSyncToString());
-    return map;
-  }
-
-  public static Optional<TableSyncMetadata> fromMap(Map<String, String> 
properties) {
-    if (properties != null) {
-      Instant lastInstantSynced = null;
-      List<Instant> instantsToConsiderForNextSync = null;
-      if (properties.containsKey(XTABLE_LAST_INSTANT_SYNCED_PROP)) {
-        lastInstantSynced = 
Instant.parse(properties.get(XTABLE_LAST_INSTANT_SYNCED_PROP));
-      }
-      if 
(properties.containsKey(INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP)) {
-        instantsToConsiderForNextSync =
-            convertStringToInstantsToConsiderForNextSync(
-                
properties.get(INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP));
-      }
-      return Optional.ofNullable(
-          TableSyncMetadata.of(lastInstantSynced, 
instantsToConsiderForNextSync));
-    }
-    return Optional.empty();
-  }
-
-  private String convertInstantsToConsiderForNextSyncToString() {
-    if (instantsToConsiderForNextSync == null || 
instantsToConsiderForNextSync.isEmpty()) {
-      return "";
-    }
-    Collections.sort(instantsToConsiderForNextSync);
-    return instantsToConsiderForNextSync.stream()
-        .map(Instant::toString)
-        .collect(Collectors.joining(","));
-  }
-
-  private static List<Instant> convertStringToInstantsToConsiderForNextSync(
-      String instantsToConsiderForNextSync) {
-    if (instantsToConsiderForNextSync == null || 
instantsToConsiderForNextSync.isEmpty()) {
-      return Collections.emptyList();
-    }
-    List<Instant> instantsList =
-        Arrays.stream(instantsToConsiderForNextSync.split(","))
-            .map(String::trim)
-            .map(Instant::parse)
-            .collect(Collectors.toList());
-    return instantsList;
-  }
-}
diff --git 
a/api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java 
b/api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java
new file mode 100644
index 00000000..fa5eeaff
--- /dev/null
+++ b/api/src/main/java/org/apache/xtable/model/metadata/TableSyncMetadata.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.model.metadata;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Value;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+
+import org.apache.xtable.model.exception.ParseException;
+
+/**
+ * Metadata representing the state of a table sync process. This metadata is 
stored in the target
+ * table's properties and is used to track the status of previous sync 
operation.
+ */
+@Value
+@AllArgsConstructor(access = AccessLevel.PRIVATE)
+public class TableSyncMetadata {
+  private static final int CURRENT_VERSION = 0;
+  private static final ObjectMapper MAPPER =
+      new ObjectMapper()
+          .registerModule(new JavaTimeModule())
+          .configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false)
+          .setSerializationInclusion(JsonInclude.Include.NON_NULL);
+
+  /** Property name for the XTABLE metadata in the table metadata/properties */
+  public static final String XTABLE_METADATA = "XTABLE_METADATA";
+
+  Instant lastInstantSynced;
+  List<Instant> instantsToConsiderForNextSync;
+  int version;
+
+  public static TableSyncMetadata of(
+      Instant lastInstantSynced, List<Instant> instantsToConsiderForNextSync) {
+    return new TableSyncMetadata(lastInstantSynced, 
instantsToConsiderForNextSync, CURRENT_VERSION);
+  }
+
+  public String toJson() {
+    try {
+      return MAPPER.writeValueAsString(this);
+    } catch (IOException e) {
+      throw new ParseException("Failed to serialize TableSyncMetadata", e);
+    }
+  }
+
+  public static Optional<TableSyncMetadata> fromJson(String metadata) {
+    if (metadata == null || metadata.isEmpty()) {
+      return Optional.empty();
+    } else {
+      try {
+        TableSyncMetadata parsedMetadata = MAPPER.readValue(metadata, 
TableSyncMetadata.class);
+        if (parsedMetadata.getLastInstantSynced() == null) {
+          throw new ParseException("LastInstantSynced is required in 
TableSyncMetadata");
+        }
+        if (parsedMetadata.getVersion() > CURRENT_VERSION) {
+          throw new ParseException(
+              "Unable handle metadata version: "
+                  + parsedMetadata.getVersion()
+                  + " max supported version: "
+                  + CURRENT_VERSION);
+        }
+        return Optional.of(parsedMetadata);
+      } catch (IOException e) {
+        throw new ParseException("Failed to deserialize TableSyncMetadata", e);
+      }
+    }
+  }
+}
diff --git a/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java 
b/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
index 2ff6fb27..a0df756a 100644
--- a/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
+++ b/api/src/main/java/org/apache/xtable/spi/sync/ConversionTarget.java
@@ -25,7 +25,7 @@ import org.apache.hadoop.conf.Configuration;
 
 import org.apache.xtable.conversion.PerTableConfig;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
 import org.apache.xtable.model.storage.DataFilesDiff;
diff --git a/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java 
b/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
index 41e91525..7cd0b384 100644
--- a/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
+++ b/api/src/main/java/org/apache/xtable/spi/sync/TableFormatSync.java
@@ -37,7 +37,7 @@ import org.apache.xtable.model.IncrementalTableChanges;
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.TableChange;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.sync.SyncMode;
 import org.apache.xtable.model.sync.SyncResult;
 
diff --git 
a/api/src/test/java/org/apache/xtable/model/metadata/TestTableSyncMetadata.java 
b/api/src/test/java/org/apache/xtable/model/metadata/TestTableSyncMetadata.java
new file mode 100644
index 00000000..3dcc6fbe
--- /dev/null
+++ 
b/api/src/test/java/org/apache/xtable/model/metadata/TestTableSyncMetadata.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+ 
+package org.apache.xtable.model.metadata;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import org.apache.xtable.model.exception.ParseException;
+
+class TestTableSyncMetadata {
+
+  @ParameterizedTest
+  @MethodSource("provideMetadataAndJson")
+  void jsonRoundTrip(TableSyncMetadata metadata, String expectedJson) {
+    assertEquals(expectedJson, metadata.toJson());
+    assertEquals(metadata, TableSyncMetadata.fromJson(expectedJson).get());
+  }
+
+  private static Stream<Arguments> provideMetadataAndJson() {
+    return Stream.of(
+        Arguments.of(
+            TableSyncMetadata.of(
+                Instant.parse("2020-07-04T10:15:30.00Z"),
+                Arrays.asList(
+                    Instant.parse("2020-08-21T11:15:30.00Z"),
+                    Instant.parse("2024-01-21T12:15:30.00Z"))),
+            
"{\"lastInstantSynced\":\"2020-07-04T10:15:30Z\",\"instantsToConsiderForNextSync\":[\"2020-08-21T11:15:30Z\",\"2024-01-21T12:15:30Z\"],\"version\":0}"),
+        Arguments.of(
+            TableSyncMetadata.of(Instant.parse("2020-07-04T10:15:30.00Z"), 
Collections.emptyList()),
+            
"{\"lastInstantSynced\":\"2020-07-04T10:15:30Z\",\"instantsToConsiderForNextSync\":[],\"version\":0}"),
+        Arguments.of(
+            TableSyncMetadata.of(Instant.parse("2020-07-04T10:15:30.00Z"), 
null),
+            "{\"lastInstantSynced\":\"2020-07-04T10:15:30Z\",\"version\":0}"));
+  }
+
+  @Test
+  void failToParseJsonFromNewerVersion() {
+    assertThrows(
+        ParseException.class,
+        () ->
+            TableSyncMetadata.fromJson(
+                
"{\"lastInstantSynced\":\"2020-07-04T10:15:30Z\",\"instantsToConsiderForNextSync\":[\"2020-08-21T11:15:30Z\",\"2024-01-21T12:15:30Z\"],\"version\":1}"));
+  }
+
+  @Test
+  void failToParseJsonWithMissingLastSyncedInstant() {
+    assertThrows(
+        ParseException.class,
+        () ->
+            TableSyncMetadata.fromJson(
+                
"{\"instantsToConsiderForNextSync\":[\"2020-08-21T11:15:30Z\",\"2024-01-21T12:15:30Z\"],\"version\":0}"));
+  }
+}
diff --git 
a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java 
b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
index d401b08e..2a9e0588 100644
--- a/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
+++ b/api/src/test/java/org/apache/xtable/spi/sync/TestTableFormatSync.java
@@ -43,7 +43,7 @@ import org.apache.xtable.model.IncrementalTableChanges;
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.TableChange;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalField;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
diff --git 
a/core/src/main/java/org/apache/xtable/conversion/ConversionController.java 
b/core/src/main/java/org/apache/xtable/conversion/ConversionController.java
index f12c8fa4..34968733 100644
--- a/core/src/main/java/org/apache/xtable/conversion/ConversionController.java
+++ b/core/src/main/java/org/apache/xtable/conversion/ConversionController.java
@@ -42,7 +42,7 @@ import org.apache.xtable.exception.ReadException;
 import org.apache.xtable.model.IncrementalTableChanges;
 import org.apache.xtable.model.InstantsForIncrementalSync;
 import org.apache.xtable.model.InternalSnapshot;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.sync.SyncMode;
 import org.apache.xtable.model.sync.SyncResult;
 import org.apache.xtable.spi.extractor.ConversionSource;
diff --git 
a/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java 
b/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
index 910fe1d7..50c6f74f 100644
--- a/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
+++ b/core/src/main/java/org/apache/xtable/delta/DeltaConversionTarget.java
@@ -57,7 +57,7 @@ import com.google.common.annotations.VisibleForTesting;
 import org.apache.xtable.conversion.PerTableConfig;
 import org.apache.xtable.exception.NotSupportedException;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
 import org.apache.xtable.model.storage.DataFilesDiff;
@@ -202,9 +202,12 @@ public class DeltaConversionTarget implements 
ConversionTarget {
 
   @Override
   public Optional<TableSyncMetadata> getTableMetadata() {
-    return TableSyncMetadata.fromMap(
-        
JavaConverters.mapAsJavaMapConverter(deltaLog.snapshot().metadata().configuration())
-            .asJava());
+    return TableSyncMetadata.fromJson(
+        deltaLog
+            .snapshot()
+            .metadata()
+            .configuration()
+            .getOrElse(TableSyncMetadata.XTABLE_METADATA, null));
   }
 
   @Override
@@ -265,9 +268,10 @@ public class DeltaConversionTarget implements 
ConversionTarget {
     }
 
     private Map<String, String> getConfigurationsForDeltaSync() {
-      Map<String, String> configMap = new HashMap<>(metadata.asMap());
+      Map<String, String> configMap = new HashMap<>();
       configMap.put(DeltaConfigs.MIN_READER_VERSION().key(), 
MIN_READER_VERSION);
       configMap.put(DeltaConfigs.MIN_WRITER_VERSION().key(), 
MIN_WRITER_VERSION);
+      configMap.put(TableSyncMetadata.XTABLE_METADATA, metadata.toJson());
       // Sets retention for the Delta Log, does not impact underlying files in 
the table
       configMap.put(
           DeltaConfigs.LOG_RETENTION().key(), String.format("interval %d 
hours", retentionInHours));
diff --git 
a/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java 
b/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
index a778462d..b1a0bc91 100644
--- a/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
+++ b/core/src/main/java/org/apache/xtable/hudi/HudiConversionTarget.java
@@ -24,7 +24,6 @@ import java.io.IOException;
 import java.time.temporal.ChronoUnit;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -84,7 +83,7 @@ import org.apache.xtable.exception.NotSupportedException;
 import org.apache.xtable.exception.ReadException;
 import org.apache.xtable.exception.UpdateException;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalField;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
@@ -308,7 +307,10 @@ public class HudiConversionTarget implements 
ConversionTarget {
                         throw new ReadException("Unable to read Hudi commit 
metadata", ex);
                       }
                     })
-                .flatMap(TableSyncMetadata::fromMap));
+                .flatMap(
+                    metadata ->
+                        TableSyncMetadata.fromJson(
+                            metadata.get(TableSyncMetadata.XTABLE_METADATA))));
   }
 
   @Override
@@ -549,7 +551,8 @@ public class HudiConversionTarget implements 
ConversionTarget {
     }
 
     private Option<Map<String, String>> getExtraMetadata() {
-      Map<String, String> extraMetadata = new 
HashMap<>(tableSyncMetadata.asMap());
+      Map<String, String> extraMetadata =
+          Collections.singletonMap(TableSyncMetadata.XTABLE_METADATA, 
tableSyncMetadata.toJson());
       return Option.of(extraMetadata);
     }
 
diff --git 
a/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java 
b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
index cb9cfeae..72125cb0 100644
--- a/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
+++ b/core/src/main/java/org/apache/xtable/iceberg/IcebergConversionTarget.java
@@ -18,13 +18,9 @@
  
 package org.apache.xtable.iceberg;
 
-import static 
org.apache.xtable.model.TableSyncMetadata.INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP;
-import static 
org.apache.xtable.model.TableSyncMetadata.XTABLE_LAST_INSTANT_SYNCED_PROP;
-
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
 
 import lombok.extern.log4j.Log4j2;
@@ -45,7 +41,7 @@ import org.apache.iceberg.exceptions.NotFoundException;
 
 import org.apache.xtable.conversion.PerTableConfig;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
 import org.apache.xtable.model.storage.DataFilesDiff;
@@ -179,9 +175,7 @@ public class IcebergConversionTarget implements 
ConversionTarget {
   @Override
   public void syncMetadata(TableSyncMetadata metadata) {
     UpdateProperties updateProperties = transaction.updateProperties();
-    for (Map.Entry<String, String> stateProperty : 
metadata.asMap().entrySet()) {
-      updateProperties.set(stateProperty.getKey(), stateProperty.getValue());
-    }
+    updateProperties.set(TableSyncMetadata.XTABLE_METADATA, metadata.toJson());
     if (!table.properties().containsKey(TableProperties.WRITE_DATA_LOCATION)) {
       // Required for a consistent write location when writing back to the 
table as Iceberg
       updateProperties.set(TableProperties.WRITE_DATA_LOCATION, basePath);
@@ -240,7 +234,7 @@ public class IcebergConversionTarget implements 
ConversionTarget {
     if (table == null) {
       return Optional.empty();
     }
-    return TableSyncMetadata.fromMap(table.properties());
+    return 
TableSyncMetadata.fromJson(table.properties().get(TableSyncMetadata.XTABLE_METADATA));
   }
 
   @Override
@@ -273,11 +267,7 @@ public class IcebergConversionTarget implements 
ConversionTarget {
           // in the subsequent sync round.
           
table.manageSnapshots().rollbackTo(currentSnapshot.parentId()).commit();
           Transaction transaction = table.newTransaction();
-          transaction
-              .updateProperties()
-              .remove(XTABLE_LAST_INSTANT_SYNCED_PROP)
-              .remove(INFLIGHT_COMMITS_TO_CONSIDER_FOR_NEXT_SYNC_PROP)
-              .commit();
+          
transaction.updateProperties().remove(TableSyncMetadata.XTABLE_METADATA).commit();
           transaction.commitTransaction();
         }
       }
diff --git 
a/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java 
b/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
index 4fe7c365..558b8ee5 100644
--- 
a/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
+++ 
b/core/src/test/java/org/apache/xtable/conversion/TestConversionController.java
@@ -49,7 +49,7 @@ import org.apache.xtable.model.InstantsForIncrementalSync;
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
 import org.apache.xtable.model.TableChange;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.storage.TableFormat;
 import org.apache.xtable.model.sync.SyncMode;
 import org.apache.xtable.model.sync.SyncResult;
diff --git 
a/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java 
b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
index a0d51b7a..f6987a88 100644
--- 
a/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
+++ 
b/core/src/test/java/org/apache/xtable/hudi/ITHudiConversionSourceTarget.java
@@ -72,7 +72,7 @@ import org.apache.hudi.metadata.HoodieMetadataFileSystemView;
 
 import org.apache.xtable.conversion.PerTableConfigImpl;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalField;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
diff --git 
a/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java 
b/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
index 8f2ccd94..0e79f74f 100644
--- 
a/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
+++ 
b/core/src/test/java/org/apache/xtable/hudi/TestHudiConversionSourceTarget.java
@@ -43,7 +43,7 @@ import org.apache.hudi.common.util.Option;
 import org.apache.xtable.avro.AvroSchemaConverter;
 import org.apache.xtable.exception.NotSupportedException;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalField;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;
diff --git a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java 
b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
index 3d479b4e..638f3a6e 100644
--- a/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
+++ b/core/src/test/java/org/apache/xtable/iceberg/TestIcebergSync.java
@@ -85,7 +85,7 @@ import org.apache.xtable.ITConversionController;
 import org.apache.xtable.conversion.PerTableConfigImpl;
 import org.apache.xtable.model.InternalSnapshot;
 import org.apache.xtable.model.InternalTable;
-import org.apache.xtable.model.TableSyncMetadata;
+import org.apache.xtable.model.metadata.TableSyncMetadata;
 import org.apache.xtable.model.schema.InternalField;
 import org.apache.xtable.model.schema.InternalPartitionField;
 import org.apache.xtable.model.schema.InternalSchema;

Reply via email to