normalize years and months part in coverting Avro interval to ISO8601 string

Author: Sagar Agarwal

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertor.java

@@ -708,13 +708,9 @@ static String handleRecordFieldType(String fieldName, GenericRecord element, Sch
           Period.ZERO
               .plusYears(((Number) getOrDefault(element, "years", 0L)).longValue())
               .plusMonths(((Number) getOrDefault(element, "months", 0L)).longValue())
-              .plusDays(((Number) getOrDefault(element, "days", 0L)).longValue());
-      /*
-       * Convert the period to a ISO-8601 period formatted String, such as P6Y3M1D.
-       * A zero period will be represented as zero days, 'P0D'.
-       * Refer to javadoc for Period#toString.
-       */
-      String periodIso8061 = period.toString();
+              .plusDays(((Number) getOrDefault(element, "days", 0L)).longValue())
+              .normalized(); // Normalize years and months
+
       java.time.Duration duration =
           java.time.Duration.ZERO
               .plusHours(((Number) getOrDefault(element, "hours", 0L)).longValue())
@@ -726,12 +722,19 @@ static String handleRecordFieldType(String fieldName, GenericRecord element, Sch
        * refer to javadoc for Duration#toString.
        */
       String durationIso8610 = duration.toString();
-      // Convert to ISO-8601 period format.
-      if (duration.isZero()) {
-        return periodIso8061;
-      } else {
-        return periodIso8061 + StringUtils.removeStartIgnoreCase(durationIso8610, "P");
+      if (period.isZero() && duration.isZero()) {
+        // Interval of length 0 is represented as P0Y.
+        return "P0Y";
+      } else if (period.isZero()) {
+        // If year-month-day part is 0, get ISO8601 formatted string from duration part.
+        return duration.toString();
+      } else if (duration.isZero()) {
+        // If hour-minute-second part is 0, get ISO8601 formatted string from period part.
+        return period.toString();
       }
+
+      // Combine both non-zero parts into ISO8601 string.
+      return period + StringUtils.removeStartIgnoreCase(durationIso8610, "P");
     } else {
       throw new UnsupportedOperationException(
           String.format(

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/avro/GenericRecordTypeConvertorTest.java

@@ -712,10 +712,9 @@ public void testIntervalNanos() {
     result =
         GenericRecordTypeConvertor.handleRecordFieldType(
             "interval_nanos_column",
-            AvroTestingHelper.createIntervalNanosRecord(1000L, 1000L, 3890L, 25L, 331L, 12L, 9L),
+            AvroTestingHelper.createIntervalNanosRecord(1000L, 1000L, 3890L, 25L, 331L, 3672L, 9L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
-    assertEquals(
-        "Test #1 interval nano conversion:", "P1000Y1000M3890DT30H31M12.000000009S", result);
+    assertEquals("Test #1 interval nano conversion:", "P1083Y4M3890DT31H32M12.000000009S", result);
 
     /* Test with any field set as null gets treated as 0. */
     result =
@@ -725,7 +724,7 @@ public void testIntervalNanos() {
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
     assertEquals(
         "Test #2 interval nano conversion with null minutes:",
-        "P1000Y1000M3890DT25H12.000000009S",
+        "P1083Y4M3890DT25H12.000000009S",
         result);
 
     /* Basic test for negative field. */
@@ -736,18 +735,28 @@ public void testIntervalNanos() {
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
     assertEquals(
         "Test #3 interval nano conversion with negative months:",
-        "P1000Y-1000M3890DT25H31M12.000000009S",
+        "P916Y8M3890DT25H31M12.000000009S",
         result);
 
+    /* Test all negative */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(
+                -1000L, -1000L, -3890L, -2L, -31L, -12L, -90L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals(
+        "Test #4 all fields are negative:", "P-1083Y-4M-3890DT-2H-31M-12.00000009S", result);
+
     /* Test that negative nanos subtract from the fractional seconds, for example 12 Seconds -1 Nanos becomes 11.999999991s. */
     result =
         GenericRecordTypeConvertor.handleRecordFieldType(
             "interval_nanos_column",
             AvroTestingHelper.createIntervalNanosRecord(1000L, 31L, 3890L, 25L, 31L, 12L, -9L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
     assertEquals(
-        "Test #4 interval nano conversion with negative nanos:",
-        "P1000Y31M3890DT25H31M11.999999991S",
+        "Test #5 interval nano conversion with negative nanos:",
+        "P1002Y7M3890DT25H31M11.999999991S",
         result);
 
     /* Test 0 interval. */
@@ -756,15 +765,15 @@ public void testIntervalNanos() {
             "interval_nanos_column",
             AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, 0L, 0L, 0L, 0L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
-    assertEquals("Test #5 interval nano conversion with all zeros", "P0D", result);
+    assertEquals("Test #6 interval nano conversion with all zeros", "P0Y", result);
 
     /* Test almost zero interval with only nanos set. */
     result =
         GenericRecordTypeConvertor.handleRecordFieldType(
             "interval_nanos_column",
             AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, 0L, 0L, 0L, 1L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
-    assertEquals("Test #6 interval nano conversion with only nanos", "P0DT0.000000001S", result);
+    assertEquals("Test #7 interval nano conversion with only nanos", "PT0.000000001S", result);
     /* Test with large values. */
     result =
         GenericRecordTypeConvertor.handleRecordFieldType(
@@ -773,7 +782,7 @@ public void testIntervalNanos() {
                 2147483647L, 11L, 2147483647L, 2147483647L, 2147483647L, 2147483647L, 999999999L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
     assertEquals(
-        "Test #6 interval nano conversion with INT.MAX values",
+        "Test #8 interval nano conversion with INT.MAX values",
         "P2147483647Y11M2147483647DT2183871564H21M7.999999999S",
         result);
 
@@ -791,9 +800,65 @@ public void testIntervalNanos() {
                 -999999999L),
             AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
     assertEquals(
-        "Test #6 interval nano conversion with -INT.MAX values",
+        "Test #9 interval nano conversion with -INT.MAX values",
         "P-2147483647Y-11M-2147483647DT-2183871564H-21M-7.999999999S",
         result);
+
+    /* Test with only year part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(1L, 0L, 0L, 0L, 0L, 0L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #10 interval nano conversion with only years", "P1Y", result);
+
+    /* Test with only month part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, -1L, 0L, 0L, 0L, 0L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #11 interval nano conversion with only months", "P-1M", result);
+
+    /* Test with only days part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 1L, 0L, 0L, 0L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #12 interval nano conversion with only days", "P1D", result);
+
+    /* Test with only hours part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, -1L, 0L, 0L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #13 interval nano conversion with only hours", "PT-1H", result);
+
+    /* Test with only minutes part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, 0L, 12L, 0L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #14 interval nano conversion with only minutes", "PT12M", result);
+
+    /* Test with only seconds part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, 0L, 0L, 1L, 0L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #15 interval nano conversion with only seconds", "PT1S", result);
+
+    /* Test with only nanos part */
+    result =
+        GenericRecordTypeConvertor.handleRecordFieldType(
+            "interval_nanos_column",
+            AvroTestingHelper.createIntervalNanosRecord(0L, 0L, 0L, 0L, 0L, 0L, 100L),
+            AvroTestingHelper.INTERVAL_NANOS_SCHEMA);
+    assertEquals("Test #16 interval nano conversion with only nanos", "PT0.0000001S", result);
   }
 
   @Test