Use time-millis for time (sec) vectors instead of time-micros (consistent with timestamp and should not roll over for time of day)

Author: martin-traverse

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/ArrowToAvroUtils.java

@@ -132,8 +132,8 @@ public class ArrowToAvroUtils {
    *   <li>ArrowType.FixedSizeBinary --> FIXED
    *   <li>ArrowType.Decimal --> decimal (FIXED)
    *   <li>ArrowType.Date --> date (INT)
-   *   <li>ArrowType.Time (MILLI) --> time-millis (INT)
-   *   <li>ArrowType.Time (SEC | MICRO | NANO) --> time-micros (LONG)
+   *   <li>ArrowType.Time (SEC | MILLI) --> time-millis (INT)
+   *   <li>ArrowType.Time (MICRO | NANO) --> time-micros (LONG)
    *   <li>ArrowType.Timestamp (NANOSECONDS, TZ != NULL) --> time-nanos (LONG)
    *   <li>ArrowType.Timestamp (MICROSECONDS, TZ != NULL) --> time-micros (LONG)
    *   <li>ArrowType.Timestamp (MILLISECONDS | SECONDS, TZ != NULL) --> time-millis (LONG)
@@ -326,7 +326,7 @@ private static <T> T buildBaseTypeSchema(
 
       case Time:
         ArrowType.Time timeType = (ArrowType.Time) field.getType();
-        if (timeType.getUnit() == TimeUnit.MILLISECOND) {
+        if ((timeType.getUnit() == TimeUnit.SECOND || timeType.getUnit() == TimeUnit.MILLISECOND)) {
           return builder.intBuilder().prop("logicalType", "time-millis").endInt();
         } else {
           // All other time types (sec, micro, nano) are encoded as time-micros (LONG)
@@ -410,7 +410,7 @@ private static <T> SchemaBuilder.FieldAssembler<T> buildBaseFieldSchema(
 
       case Time:
         ArrowType.Time timeType = (ArrowType.Time) field.getType();
-        if (timeType.getUnit() == TimeUnit.MILLISECOND) {
+        if ((timeType.getUnit() == TimeUnit.SECOND || timeType.getUnit() == TimeUnit.MILLISECOND)) {
           return builder.intBuilder().prop("logicalType", "time-millis").endInt().noDefault();
         } else {
           // All other time types (sec, micro, nano) are encoded as time-micros (LONG)
@@ -504,7 +504,7 @@ private static <T> SchemaBuilder.FieldAssembler<T> buildBaseFieldSchema(
 
       case Time:
         ArrowType.Time timeType = (ArrowType.Time) field.getType();
-        if (timeType.getUnit() == TimeUnit.MILLISECOND) {
+        if ((timeType.getUnit() == TimeUnit.SECOND || timeType.getUnit() == TimeUnit.MILLISECOND)) {
           return (SchemaBuilder.UnionAccumulator)
               builder.intBuilder().prop("logicalType", "time-millis").endInt();
         } else {

adapter/avro/src/main/java/org/apache/arrow/adapter/avro/producers/logical/AvroTimeSecProducer.java

@@ -27,10 +27,12 @@
  */
 public class AvroTimeSecProducer extends BaseAvroProducer<TimeSecVector> {
 
-  // Convert seconds to microseconds for Avro time-micros (LONG) type
-  // Range is 1000 times more than for milliseconds, so won't fit into time-millis (INT)
+  // Convert seconds to milliseconds for Avro time-millis (INT) type
+  // INT is enough to cover the number of milliseconds in a day
+  // So overflows should not happen if values are valid times of day
 
-  private static final long MICROS_PER_SECOND = 1000000;
+  private static final int MILLIS_PER_SECOND = 1000;
+  private static final long OVERFLOW_LIMIT = Integer.MAX_VALUE / 1000;
 
   /** Instantiate an AvroTimeSecProducer. */
   public AvroTimeSecProducer(TimeSecVector vector) {
@@ -40,8 +42,11 @@ public AvroTimeSecProducer(TimeSecVector vector) {
   @Override
   public void produce(Encoder encoder) throws IOException {
     int seconds = vector.getDataBuffer().getInt(currentIndex * (long) TimeSecVector.TYPE_WIDTH);
-    long micros = seconds * MICROS_PER_SECOND;
-    encoder.writeLong(micros);
+    if (Math.abs(seconds) > OVERFLOW_LIMIT) {
+      throw new ArithmeticException("Time value is too large for Avro encoding");
+    }
+    int millis = seconds * MILLIS_PER_SECOND;
+    encoder.writeInt(millis);
     currentIndex++;
   }
 }

adapter/avro/src/test/java/org/apache/arrow/adapter/avro/ArrowToAvroDataTest.java

@@ -1240,7 +1240,7 @@ public void testWriteTimes() throws Exception {
         // Read and check values
         for (int row = 0; row < rowCount; row++) {
           record = datumReader.read(record, decoder);
-          assertEquals(timeSecVector.get(row), (int) ((long) record.get("timeSec") / 1000000));
+          assertEquals(timeSecVector.get(row), (int) (record.get("timeSec")) / 1000);
           assertEquals(timeMillisVector.get(row), record.get("timeMillis"));
           assertEquals(timeMicrosVector.get(row), record.get("timeMicros"));
           assertEquals(timeNanosVector.get(row), (long) record.get("timeNanos") * 1000);
@@ -1321,7 +1321,7 @@ public void testWriteNullableTimes() throws Exception {
 
         for (int row = 1; row < rowCount; row++) {
           record = datumReader.read(record, decoder);
-          assertEquals(timeSecVector.get(row), (int) ((long) record.get("timeSec") / 1000000));
+          assertEquals(timeSecVector.get(row), ((int) record.get("timeSec") / 1000));
           assertEquals(timeMillisVector.get(row), record.get("timeMillis"));
           assertEquals(timeMicrosVector.get(row), record.get("timeMicros"));
           assertEquals(timeNanosVector.get(row), (long) record.get("timeNanos") * 1000);

adapter/avro/src/test/java/org/apache/arrow/adapter/avro/ArrowToAvroSchemaTest.java

@@ -515,15 +515,15 @@ public void testConvertTimeTypes() {
     assertEquals(Schema.Type.UNION, schema.getField("nullableTimeSec").schema().getType());
     assertEquals(2, schema.getField("nullableTimeSec").schema().getTypes().size());
     Schema nullableTimeSecSchema = schema.getField("nullableTimeSec").schema().getTypes().get(0);
-    assertEquals(Schema.Type.LONG, nullableTimeSecSchema.getType());
-    assertEquals("time-micros", nullableTimeSecSchema.getProp("logicalType"));
+    assertEquals(Schema.Type.INT, nullableTimeSecSchema.getType());
+    assertEquals("time-millis", nullableTimeSecSchema.getProp("logicalType"));
     assertEquals(
         Schema.Type.NULL, schema.getField("nullableTimeSec").schema().getTypes().get(1).getType());
 
     // Assertions for nonNullableTimeSec
     Schema nonNullableTimeSecSchema = schema.getField("nonNullableTimeSec").schema();
-    assertEquals(Schema.Type.LONG, nonNullableTimeSecSchema.getType());
-    assertEquals("time-micros", nonNullableTimeSecSchema.getProp("logicalType"));
+    assertEquals(Schema.Type.INT, nonNullableTimeSecSchema.getType());
+    assertEquals("time-millis", nonNullableTimeSecSchema.getProp("logicalType"));
 
     // Assertions for nullableTimeMillis
     assertEquals(Schema.Type.UNION, schema.getField("nullableTimeMillis").schema().getType());