[AVRO-4056][Java] Add time-nanos logical type

lang/java/avro/src/main/java/org/apache/avro/LogicalTypes.java

@@ -158,6 +158,9 @@ private static LogicalType fromSchemaImpl(Schema schema, boolean throwErrors) {
       case TIME_MICROS:
         logicalType = TIME_MICROS_TYPE;
         break;
+      case TIME_NANOS:
+        logicalType = TIME_NANOS_TYPE;
+        break;
       case LOCAL_TIMESTAMP_MICROS:
         logicalType = LOCAL_TIMESTAMP_MICROS_TYPE;
         break;
@@ -197,6 +200,7 @@ private static LogicalType fromSchemaImpl(Schema schema, boolean throwErrors) {
   private static final String DATE = "date";
   private static final String TIME_MILLIS = "time-millis";
   private static final String TIME_MICROS = "time-micros";
+  private static final String TIME_NANOS = "time-nanos";
   private static final String TIMESTAMP_MILLIS = "timestamp-millis";
   private static final String TIMESTAMP_MICROS = "timestamp-micros";
   private static final String TIMESTAMP_NANOS = "timestamp-nanos";
@@ -251,6 +255,12 @@ public static TimeMicros timeMicros() {
     return TIME_MICROS_TYPE;
   }
 
+  private static final TimeNanos TIME_NANOS_TYPE = new TimeNanos();
+
+  public static TimeNanos timeNanos() {
+    return TIME_NANOS_TYPE;
+  }
+
   private static final TimestampMillis TIMESTAMP_MILLIS_TYPE = new TimestampMillis();
 
   public static TimestampMillis timestampMillis() {
@@ -498,6 +508,21 @@ public void validate(Schema schema) {
     }
   }
 
+  /** TimeNanos represents a time in nanoseconds without a date */
+  public static class TimeNanos extends LogicalType {
+    private TimeNanos() {
+      super(TIME_NANOS);
+    }
+
+    @Override
+    public void validate(Schema schema) {
+      super.validate(schema);
+      if (schema.getType() != Schema.Type.LONG) {
+        throw new IllegalArgumentException("Time (nanos) can only be used with an underlying long type");
+      }
+    }
+  }
+
   /** TimestampMillis represents a date and time in milliseconds */
   public static class TimestampMillis extends LogicalType {
     private TimestampMillis() {

lang/java/avro/src/main/java/org/apache/avro/data/TimeConversions.java

@@ -125,6 +125,38 @@ public Schema getRecommendedSchema() {
     }
   }
 
+  public static class TimeNanosConversion extends Conversion<LocalTime> {
+    @Override
+    public Class<LocalTime> getConvertedType() {
+      return LocalTime.class;
+    }
+
+    @Override
+    public String getLogicalTypeName() {
+      return "time-nanos";
+    }
+
+    @Override
+    public String adjustAndSetValue(String varName, String valParamName) {
+      return varName + " = " + valParamName + ".truncatedTo(java.time.temporal.ChronoUnit.NANOS);";
+    }
+
+    @Override
+    public LocalTime fromLong(Long nanosFromMidnight, Schema schema, LogicalType type) {
+      return LocalTime.ofNanoOfDay(TimeUnit.NANOSECONDS.toNanos(nanosFromMidnight));
+    }
+
+    @Override
+    public Long toLong(LocalTime time, Schema schema, LogicalType type) {
+      return TimeUnit.NANOSECONDS.toNanos(time.toNanoOfDay());
+    }
+
+    @Override
+    public Schema getRecommendedSchema() {
+      return LogicalTypes.timeNanos().addToSchema(Schema.create(Schema.Type.LONG));
+    }
+  }
+
   public static class TimestampMillisConversion extends Conversion<Instant> {
     @Override
     public Class<Instant> getConvertedType() {

lang/java/avro/src/test/java/org/apache/avro/data/TestTimeConversions.java

@@ -31,6 +31,7 @@
 import org.apache.avro.data.TimeConversions.DateConversion;
 import org.apache.avro.data.TimeConversions.TimeMicrosConversion;
 import org.apache.avro.data.TimeConversions.TimeMillisConversion;
+import org.apache.avro.data.TimeConversions.TimeNanosConversion;
 import org.apache.avro.data.TimeConversions.TimestampMicrosConversion;
 import org.apache.avro.data.TimeConversions.TimestampMillisConversion;
 import org.apache.avro.reflect.ReflectData;
@@ -42,6 +43,7 @@ public class TestTimeConversions {
   public static Schema DATE_SCHEMA;
   public static Schema TIME_MILLIS_SCHEMA;
   public static Schema TIME_MICROS_SCHEMA;
+  public static Schema TIME_NANOS_SCHEMA;
   public static Schema TIMESTAMP_MILLIS_SCHEMA;
   public static Schema TIMESTAMP_MICROS_SCHEMA;
 
@@ -50,6 +52,7 @@ public static void createSchemas() {
     TestTimeConversions.DATE_SCHEMA = LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT));
     TestTimeConversions.TIME_MILLIS_SCHEMA = LogicalTypes.timeMillis().addToSchema(Schema.create(Schema.Type.INT));
     TestTimeConversions.TIME_MICROS_SCHEMA = LogicalTypes.timeMicros().addToSchema(Schema.create(Schema.Type.LONG));
+    TestTimeConversions.TIME_NANOS_SCHEMA = LogicalTypes.timeNanos().addToSchema(Schema.create(Schema.Type.LONG));
     TestTimeConversions.TIMESTAMP_MILLIS_SCHEMA = LogicalTypes.timestampMillis()
         .addToSchema(Schema.create(Schema.Type.LONG));
     TestTimeConversions.TIMESTAMP_MICROS_SCHEMA = LogicalTypes.timestampMicros()
@@ -116,6 +119,28 @@ void timeMicrosConversion() throws Exception {
         "15:14:15.926551 should be " + afternoonMicros);
   }
 
+  @Test
+  void timeNanosConversion() throws Exception {
+    TimeNanosConversion conversion = new TimeNanosConversion();
+    LocalTime oneAM = LocalTime.of(1, 0);
+    LocalTime afternoon = LocalTime.of(15, 14, 15, 926_551_123);
+    long afternoonNanos = ((long) (15 * 60 + 14) * 60 + 15) * 1_000_000_000 + 926_551_123;
+
+    assertEquals(LocalTime.MIDNIGHT, conversion.fromLong(0L, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "Midnight should be 0");
+    assertEquals(oneAM, conversion.fromLong(3_600_000_000_000L, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "01:00 should be 3,600,000,000,000");
+    assertEquals(afternoon, conversion.fromLong(afternoonNanos, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "15:14:15.926551123 should be " + afternoonNanos);
+
+    assertEquals(0, (long) conversion.toLong(LocalTime.MIDNIGHT, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "Midnight should be 0");
+    assertEquals(3_600_000_000_000L, (long) conversion.toLong(oneAM, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "01:00 should be 3,600,000,000,000");
+    assertEquals(afternoonNanos, (long) conversion.toLong(afternoon, TIME_NANOS_SCHEMA, LogicalTypes.timeNanos()),
+        "15:14:15.926551123 should be " + afternoonNanos);
+  }
+
   @Test
   void timestampMillisConversion() throws Exception {
     TimestampMillisConversion conversion = new TimestampMillisConversion();
@@ -209,6 +234,12 @@ void dynamicSchemaWithTimeMicrosConversion() throws ClassNotFoundException {
     assertEquals(TIME_MICROS_SCHEMA, schema, "Reflected schema should be logicalType timeMicros");
   }
 
+  @Test
+  void dynamicSchemaWithTimeNanosConversion() throws ClassNotFoundException {
+    Schema schema = getReflectedSchemaByName("java.time.LocalTime", new TimeConversions.TimeNanosConversion());
+    assertEquals(TIME_NANOS_SCHEMA, schema, "Reflected schema should be logicalType timeNanos");
+  }
+
   @Test
   void dynamicSchemaWithDateTimeConversion() throws ClassNotFoundException {
     Schema schema = getReflectedSchemaByName("java.time.Instant", new TimeConversions.TimestampMillisConversion());

lang/java/avro/src/test/java/org/apache/avro/reflect/TestReflectLogicalTypes.java

@@ -65,6 +65,7 @@ public class TestReflectLogicalTypes {
   public static void addUUID() {
     REFLECT.addLogicalTypeConversion(new Conversions.UUIDConversion());
     REFLECT.addLogicalTypeConversion(new Conversions.DecimalConversion());
+    REFLECT.addLogicalTypeConversion(new TimeConversions.TimeNanosConversion());
     REFLECT.addLogicalTypeConversion(new TimeConversions.LocalTimestampMillisConversion());
   }