GH-50142: [C++][Python] Honor Parquet time coercion and format version

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -1873,7 +1873,8 @@ void MakeDateTimeTypesTable(std::shared_ptr<Table>* out, bool expected_micro = f
   auto f3_x = field("f3", ::arrow::timestamp(TimeUnit::MICRO));
   auto f4 = field("f4", ::arrow::time32(TimeUnit::MILLI));
   auto f5 = field("f5", ::arrow::time64(TimeUnit::MICRO));
-  auto f6 = field("f6", ::arrow::time64(TimeUnit::NANO));
+  auto f6 =
+      field("f6", ::arrow::time64(expected_micro ? TimeUnit::MICRO : TimeUnit::NANO));
 
   std::shared_ptr<::arrow::Schema> schema(
       new ::arrow::Schema({f0, f1, f2, (expected_micro ? f3_x : f3), f4, f5, f6}));
@@ -1899,7 +1900,8 @@ void MakeDateTimeTypesTable(std::shared_ptr<Table>* out, bool expected_micro = f
                                                    &a3_x);
   ArrayFromVector<::arrow::Time32Type, int32_t>(f4->type(), is_valid, t32_values, &a4);
   ArrayFromVector<::arrow::Time64Type, int64_t>(f5->type(), is_valid, t64_us_values, &a5);
-  ArrayFromVector<::arrow::Time64Type, int64_t>(f6->type(), is_valid, t64_ns_values, &a6);
+  ArrayFromVector<::arrow::Time64Type, int64_t>(
+      f6->type(), is_valid, expected_micro ? t64_us_values : t64_ns_values, &a6);
 
   *out = Table::Make(schema, {a0, a1, a2, expected_micro ? a3_x : a3, a4, a5, a6});
 }
@@ -1927,6 +1929,35 @@ TEST(TestArrowReadWrite, DateTimeTypes) {
       CheckConfiguredRoundtrip(table, expected, parquet_version_2_4_properties));
 }
 
+TEST(TestArrowReadWrite, WriteUnannotatedTimeArrays) {
+  for (bool wide : {false, true}) {
+    ASSERT_OK_AND_ASSIGN(auto sink, ::arrow::io::BufferOutputStream::Create());
+    auto node = PrimitiveNode::Make("time", Repetition::OPTIONAL,
+                                    wide ? ParquetType::INT64 : ParquetType::INT32);
+    auto schema = std::static_pointer_cast<GroupNode>(
+        GroupNode::Make("schema", Repetition::REQUIRED, {node}));
+    auto array = ArrayFromJSON(
+        wide ? ::arrow::time64(TimeUnit::NANO) : ::arrow::time32(TimeUnit::SECOND),
+        "[1000]");
+    auto properties = default_arrow_writer_properties();
+    ArrowWriteContext ctx(::arrow::default_memory_pool(), properties.get());
+    auto writer = ParquetFileWriter::Open(sink, schema);
+    auto row_group = writer->AppendRowGroup();
+    const int16_t def_level = 1;
+    ASSERT_OK(
+        row_group->NextColumn()->WriteArrow(&def_level, nullptr, 1, *array, &ctx, true));
+    row_group->Close();
+    writer->Close();
+    ASSERT_OK_AND_ASSIGN(auto buffer, sink->Finish());
+    ASSERT_OK_AND_ASSIGN(auto reader, OpenFile(std::make_shared<BufferReader>(buffer),
+                                               ::arrow::default_memory_pool()));
+    ASSERT_OK_AND_ASSIGN(auto result, reader->ReadTable());
+    auto expected = ArrayFromJSON(wide ? ::arrow::int64() : ::arrow::int32(),
+                                  wide ? "[1000]" : "[1000000]");
+    ::arrow::AssertArraysEqual(*expected, *result->column(0)->chunk(0));
+  }
+}
+
 TEST(TestArrowReadWrite, UseDeprecatedInt96) {
   using ::arrow::ArrayFromVector;
   using ::arrow::field;

cpp/src/parquet/arrow/arrow_schema_test.cc

@@ -1871,6 +1871,38 @@ TEST_F(TestConvertArrowSchema, ParquetFlatDecimals) {
   ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(parquet_fields));
 }
 
+TEST_F(TestConvertArrowSchema, ParquetTimeCoercion) {
+  auto schema =
+      ::arrow::schema({::arrow::field("seconds", ::arrow::time32(TimeUnit::SECOND)),
+                       ::arrow::field("nanos", ::arrow::time64(TimeUnit::NANO))});
+  auto arrow_properties = ArrowWriterProperties::Builder()
+                              .coerce_timestamps(TimeUnit::NANO)
+                              ->set_time_adjusted_to_utc(true)
+                              ->build();
+  for (auto version : {ParquetVersion::PARQUET_1_0, ParquetVersion::PARQUET_2_4,
+                       ParquetVersion::PARQUET_2_6}) {
+    auto properties = WriterProperties::Builder().version(version)->build();
+    auto status =
+        ToParquetSchema(schema.get(), *properties, *arrow_properties, &result_schema_);
+    if (version != ParquetVersion::PARQUET_2_6) {
+      ASSERT_RAISES(NotImplemented, status);
+      continue;
+    }
+    ASSERT_OK(status);
+    const auto logical_type = LogicalType::Time(true, LogicalType::TimeUnit::NANOS);
+    for (int i = 0; i < schema->num_fields(); ++i) {
+      const auto* column = result_schema_->Column(i);
+      ASSERT_EQ(column->physical_type(), ParquetType::INT64);
+      ASSERT_TRUE(column->logical_type()->Equals(*logical_type));
+    }
+  }
+  arrow_properties =
+      ArrowWriterProperties::Builder().coerce_timestamps(TimeUnit::SECOND)->build();
+  ASSERT_RAISES(NotImplemented,
+                ToParquetSchema(schema.get(), *default_writer_properties(),
+                                *arrow_properties, &result_schema_));
+}
+
 TEST_F(TestConvertArrowSchema, ParquetTimeAdjustedToUTC) {
   // Verify Parquet Time types have the appropriate isAdjustedToUTC value, depending
   // on the return value of ArrowWriterProperties::write_time_adjusted_to_utc()

cpp/src/parquet/arrow/schema.cc

@@ -204,6 +204,48 @@ static std::shared_ptr<const LogicalType> TimestampLogicalTypeFromArrowTimestamp
   return LogicalType::None();
 }
 
+static Status GetTimeMetadata(const ::arrow::TimeType& type,
+                              const WriterProperties& properties,
+                              const ArrowWriterProperties& arrow_properties,
+                              ParquetType::type* physical_type,
+                              std::shared_ptr<const LogicalType>* logical_type) {
+  const bool coerce = arrow_properties.coerce_timestamps_enabled();
+  auto unit = coerce ? arrow_properties.coerce_timestamps_unit() : type.unit();
+  const bool supports_nanos = properties.version() == ParquetVersion::PARQUET_2_6;
+  if (!coerce) {
+    if (unit == ::arrow::TimeUnit::SECOND) {
+      unit = ::arrow::TimeUnit::MILLI;
+    } else if (unit == ::arrow::TimeUnit::NANO && !supports_nanos) {
+      unit = ::arrow::TimeUnit::MICRO;
+    }
+  }
+
+  *physical_type =
+      unit == ::arrow::TimeUnit::MILLI ? ParquetType::INT32 : ParquetType::INT64;
+  LogicalType::TimeUnit::unit parquet_unit;
+  switch (unit) {
+    case ::arrow::TimeUnit::MILLI:
+      parquet_unit = LogicalType::TimeUnit::MILLIS;
+      break;
+    case ::arrow::TimeUnit::MICRO:
+      parquet_unit = LogicalType::TimeUnit::MICROS;
+      break;
+    case ::arrow::TimeUnit::NANO:
+      if (!supports_nanos) {
+        return Status::NotImplemented("Nanosecond times require Parquet version 2.6");
+      }
+      parquet_unit = LogicalType::TimeUnit::NANOS;
+      break;
+    case ::arrow::TimeUnit::SECOND:
+      return Status::NotImplemented("Parquet does not support times in seconds");
+    default:
+      return Status::NotImplemented("Unsupported Arrow time unit");
+  }
+  *logical_type =
+      LogicalType::Time(arrow_properties.write_time_adjusted_to_utc(), parquet_unit);
+  return Status::OK();
+}
+
 static Status GetTimestampMetadata(const ::arrow::TimestampType& type,
                                    const WriterProperties& properties,
                                    const ArrowWriterProperties& arrow_properties,
@@ -423,24 +465,11 @@ Status FieldToNode(const std::string& name, const std::shared_ptr<Field>& field,
                                properties, arrow_properties, &type, &logical_type));
       break;
     case ArrowTypeId::TIME32:
-      type = ParquetType::INT32;
-      logical_type = LogicalType::Time(
-          /*is_adjusted_to_utc=*/arrow_properties.write_time_adjusted_to_utc(),
-          LogicalType::TimeUnit::MILLIS);
+    case ArrowTypeId::TIME64:
+      RETURN_NOT_OK(
+          GetTimeMetadata(checked_cast<const ::arrow::TimeType&>(*field->type()),
+                          properties, arrow_properties, &type, &logical_type));
       break;
-    case ArrowTypeId::TIME64: {
-      type = ParquetType::INT64;
-      auto time_type = static_cast<::arrow::Time64Type*>(field->type().get());
-      if (time_type->unit() == ::arrow::TimeUnit::NANO) {
-        logical_type = LogicalType::Time(
-            /*is_adjusted_to_utc=*/arrow_properties.write_time_adjusted_to_utc(),
-            LogicalType::TimeUnit::NANOS);
-      } else {
-        logical_type = LogicalType::Time(
-            /*is_adjusted_to_utc=*/arrow_properties.write_time_adjusted_to_utc(),
-            LogicalType::TimeUnit::MICROS);
-      }
-    } break;
     case ArrowTypeId::DURATION:
       type = ParquetType::INT64;
       break;

cpp/src/parquet/column_writer.cc

@@ -20,6 +20,7 @@
 #include <algorithm>
 #include <cstdint>
 #include <cstring>
+#include <limits>
 #include <map>
 #include <memory>
 #include <span>
@@ -45,6 +46,7 @@
 #include "arrow/util/key_value_metadata.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/rle_encoding_internal.h"
+#include "arrow/util/time.h"
 #include "arrow/util/type_traits.h"
 #include "arrow/util/unreachable.h"
 #include "arrow/visit_array_inline.h"
@@ -1546,6 +1548,11 @@ class TypedColumnWriterImpl : public ColumnWriterImpl,
     return Status::OK();
   }
 
+  template <typename ArrowType>
+  Status WriteArrowTimes(const int16_t* def_levels, const int16_t* rep_levels,
+                         int64_t num_levels, const ::arrow::Array& array,
+                         ArrowWriteContext* ctx, bool maybe_parent_nulls);
+
   Status WriteArrowTimestamps(const int16_t* def_levels, const int16_t* rep_levels,
                               int64_t num_levels, const ::arrow::Array& values,
                               ArrowWriteContext* ctx, bool maybe_parent_nulls) {
@@ -2230,21 +2237,95 @@ struct SerializeFunctor<
   }
 };
 
-template <>
-struct SerializeFunctor<Int32Type, ::arrow::Time32Type> {
-  Status Serialize(const ::arrow::Time32Array& array, ArrowWriteContext*, int32_t* out) {
-    const int32_t* input = array.raw_values();
-    const auto& type = static_cast<const ::arrow::Time32Type&>(*array.type());
-    if (type.unit() == ::arrow::TimeUnit::SECOND) {
-      for (int64_t i = 0; i < array.length(); i++) {
-        out[i] = input[i] * 1000;
+template <typename ParquetType>
+template <typename ArrowType>
+Status TypedColumnWriterImpl<ParquetType>::WriteArrowTimes(
+    const int16_t* def_levels, const int16_t* rep_levels, int64_t num_levels,
+    const ::arrow::Array& array, ArrowWriteContext* ctx, bool maybe_parent_nulls) {
+  using ArrayType = typename ::arrow::TypeTraits<ArrowType>::ArrayType;
+  const auto& values = checked_cast<const ArrayType&>(array);
+  const auto& source_type = checked_cast<const ::arrow::TimeType&>(*array.type());
+  const auto& logical_type = this->descr()->logical_type();
+  auto target_unit = source_type.unit() == ::arrow::TimeUnit::SECOND
+                         ? ::arrow::TimeUnit::MILLI
+                         : source_type.unit();
+  if (logical_type->is_time()) {
+    switch (checked_cast<const TimeLogicalType&>(*logical_type).time_unit()) {
+      case LogicalType::TimeUnit::MILLIS:
+        target_unit = ::arrow::TimeUnit::MILLI;
+        break;
+      case LogicalType::TimeUnit::MICROS:
+        target_unit = ::arrow::TimeUnit::MICRO;
+        break;
+      case LogicalType::TimeUnit::NANOS:
+        target_unit = ::arrow::TimeUnit::NANO;
+        break;
+      default:
+        return Status::Invalid("Unsupported Parquet time unit");
+    }
+  } else if constexpr (sizeof(T) != sizeof(typename ArrowType::c_type)) {
+    ARROW_UNSUPPORTED();
+  }
+  if (source_type.unit() == target_unit) {
+    return WriteArrowZeroCopy(def_levels, rep_levels, num_levels, array, ctx,
+                              maybe_parent_nulls);
+  }
+
+  const auto target_type = target_unit == ::arrow::TimeUnit::MILLI
+                               ? ::arrow::time32(target_unit)
+                               : ::arrow::time64(target_unit);
+  const auto [operation, factor] =
+      ::arrow::util::GetTimestampConversion(source_type.unit(), target_unit);
+  const bool allow_truncation = ctx->properties->coerce_timestamps_enabled() &&
+                                ctx->properties->truncated_timestamps_allowed();
+  T* out = nullptr;
+  RETURN_NOT_OK(ctx->GetScratchData<T>(array.length(), &out));
+  int64_t level_index = 0;
+  for (int64_t i = 0; i < array.length(); ++i) {
+    out[i] = 0;
+    bool valid = array.IsValid(i);
+    if (maybe_parent_nulls) {
+      // Definition levels also mask children beneath null parents.
+      while (level_index < num_levels &&
+             def_levels[level_index] < level_info_.repeated_ancestor_def_level) {
+        ++level_index;
       }
-    } else {
-      std::copy(input, input + array.length(), out);
+      valid =
+          level_index < num_levels && def_levels[level_index++] == level_info_.def_level;
+    }
+    if (!valid) {
+      continue;
+    }
+    const int64_t value = values.Value(i);
+    int64_t converted;
+    if (operation == ::arrow::util::DIVIDE) {
+      if (!allow_truncation && value % factor != 0) {
+        return Status::Invalid("Casting from ", source_type.ToString(), " to ",
+                               target_type->ToString(), " would lose data: ", value);
+      }
+      converted = value / factor;
+    } else if (::arrow::internal::MultiplyWithOverflowGeneric(value, factor,
+                                                              &converted)) {
+      return Status::Invalid("Integer overflow when casting time value ", value);
     }
-    return Status::OK();
+    if (converted < std::numeric_limits<T>::min() ||
+        converted > std::numeric_limits<T>::max()) {
+      return Status::Invalid("Integer overflow when casting time value ", value);
+    }
+    out[i] = static_cast<T>(converted);
   }
-};
+
+  const bool no_nulls =
+      this->descr()->schema_node()->is_required() || array.null_count() == 0;
+  if (!maybe_parent_nulls && no_nulls) {
+    PARQUET_CATCH_NOT_OK(WriteBatchInternal(num_levels, def_levels, rep_levels, out));
+  } else {
+    PARQUET_CATCH_NOT_OK(WriteBatchSpacedInternal(num_levels, def_levels, rep_levels,
+                                                  array.null_bitmap_data(),
+                                                  array.offset(), out));
+  }
+  return Status::OK();
+}
 
 template <>
 Status TypedColumnWriterImpl<Int32Type>::WriteArrowDense(
@@ -2263,7 +2344,12 @@ Status TypedColumnWriterImpl<Int32Type>::WriteArrowDense(
       WRITE_ZERO_COPY_CASE(INT32)
       WRITE_ZERO_COPY_CASE(DATE32)
       WRITE_SERIALIZE_CASE(DATE64)
-      WRITE_SERIALIZE_CASE(TIME32)
+    case ::arrow::Type::TIME32:
+      return WriteArrowTimes<::arrow::Time32Type>(def_levels, rep_levels, num_levels,
+                                                  array, ctx, maybe_parent_nulls);
+    case ::arrow::Type::TIME64:
+      return WriteArrowTimes<::arrow::Time64Type>(def_levels, rep_levels, num_levels,
+                                                  array, ctx, maybe_parent_nulls);
       WRITE_SERIALIZE_CASE(DECIMAL32)
       WRITE_SERIALIZE_CASE(DECIMAL64)
       WRITE_SERIALIZE_CASE(DECIMAL128)
@@ -2441,7 +2527,12 @@ Status TypedColumnWriterImpl<Int64Type>::WriteArrowDense(
       WRITE_ZERO_COPY_CASE(INT64)
       WRITE_SERIALIZE_CASE(UINT32)
       WRITE_SERIALIZE_CASE(UINT64)
-      WRITE_ZERO_COPY_CASE(TIME64)
+    case ::arrow::Type::TIME32:
+      return WriteArrowTimes<::arrow::Time32Type>(def_levels, rep_levels, num_levels,
+                                                  array, ctx, maybe_parent_nulls);
+    case ::arrow::Type::TIME64:
+      return WriteArrowTimes<::arrow::Time64Type>(def_levels, rep_levels, num_levels,
+                                                  array, ctx, maybe_parent_nulls);
       WRITE_ZERO_COPY_CASE(DURATION)
       WRITE_SERIALIZE_CASE(DECIMAL32)
       WRITE_SERIALIZE_CASE(DECIMAL64)

cpp/src/parquet/properties.h

@@ -1392,24 +1392,26 @@ class PARQUET_EXPORT ArrowWriterProperties {
       return this;
     }
 
-    /// \brief Coerce all timestamps to the specified time unit.
+    /// \brief Coerce all timestamps and times to the specified time unit.
     /// \param unit time unit to truncate to.
-    /// For Parquet versions 1.0 and 2.4, nanoseconds are casted to microseconds.
+    /// Parquet versions 1.0 and 2.4 support milliseconds and microseconds.
+    /// Parquet version 2.6 also supports nanoseconds.
     Builder* coerce_timestamps(::arrow::TimeUnit::type unit) {
       coerce_timestamps_enabled_ = true;
       coerce_timestamps_unit_ = unit;
       return this;
     }
 
-    /// \brief Allow loss of data when truncating timestamps.
+    /// \brief Allow loss of data when explicitly coercing timestamps or times.
     ///
     /// This is disallowed by default and an error will be returned.
+    /// Only applies when coerce_timestamps is explicitly enabled.
     Builder* allow_truncated_timestamps() {
       truncated_timestamps_allowed_ = true;
       return this;
     }
 
-    /// \brief Disallow loss of data when truncating timestamps (default).
+    /// \brief Disallow loss of data when truncating timestamps or times (default).
     Builder* disallow_truncated_timestamps() {
       truncated_timestamps_allowed_ = false;
       return this;