GH-50915: [FORMAT] Allow TIMESTAMP logical type to annotate FIXED_LEN_BYTE_ARRAY(12)

cpp/src/parquet/arrow/arrow_reader_writer_test.cc

@@ -2166,6 +2166,72 @@ TEST(TestArrowReadWrite, CoerceTimestampsLosePrecision) {
                                 allow_truncation_to_micros));
 }
 
+TEST(TestArrowReadWrite, FlbaTimestampConversionValues) {
+  auto node =
+      PrimitiveNode::Make("ts", Repetition::REQUIRED,
+                          LogicalType::Timestamp(true, LogicalType::TimeUnit::MICROS),
+                          ParquetType::FIXED_LEN_BYTE_ARRAY, /*length=*/12);
+  auto file_schema = std::static_pointer_cast<GroupNode>(
+      GroupNode::Make("schema", Repetition::REQUIRED, {node}));
+
+  // Little-endian 96-bit values: 1,000,000 (fits int64) and 2^64 (overflows int64).
+  uint8_t in_range[12] = {0x40, 0x42, 0x0f, 0, 0, 0, 0, 0, 0, 0, 0, 0};
+  uint8_t overflow[12] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0};
+  FLBA values[2] = {FLBA(in_range), FLBA(overflow)};
+
+  auto sink = CreateOutputStream();
+  auto writer = ParquetFileWriter::Open(sink, file_schema);
+  RowGroupWriter* rg_writer = writer->AppendRowGroup();
+  auto* col_writer = dynamic_cast<TypedColumnWriter<FLBAType>*>(rg_writer->NextColumn());
+  ASSERT_NE(col_writer, nullptr);
+  col_writer->WriteBatch(2, nullptr, nullptr, values);
+  col_writer->Close();
+  rg_writer->Close();
+  writer->Close();
+  ASSERT_OK_AND_ASSIGN(auto buffer, sink->Finish());
+
+  auto read_table = [&buffer](ArrowReaderProperties props,
+                              std::shared_ptr<Table>* out) -> ::arrow::Status {
+    FileReaderBuilder builder;
+    RETURN_NOT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+    std::unique_ptr<FileReader> reader;
+    RETURN_NOT_OK(builder.properties(props)->Build(&reader));
+    return reader->ReadTable(out);
+  };
+
+  // Default: raw, lossless FixedSizeBinary(12).
+  {
+    std::shared_ptr<Table> table;
+    ASSERT_OK(read_table(ArrowReaderProperties(), &table));
+    ASSERT_EQ(::arrow::Type::FIXED_SIZE_BINARY, table->schema()->field(0)->type()->id());
+  }
+
+  // Convert, error on overflow (default policy): the 2^64 row fails the read.
+  {
+    ArrowReaderProperties props;
+    props.set_convert_flba_timestamps(true);
+    std::shared_ptr<Table> table;
+    ASSERT_RAISES(Invalid, read_table(props, &table));
+  }
+
+  // Convert, clamp on overflow: in-range value is exact; overflow clamps to
+  // INT64_MAX.
+  {
+    ArrowReaderProperties props;
+    props.set_convert_flba_timestamps(true);
+    props.set_flba_timestamp_clamp_on_overflow(true);
+    std::shared_ptr<Table> table;
+    ASSERT_OK(read_table(props, &table));
+    ASSERT_EQ(*::arrow::timestamp(TimeUnit::MICRO, "UTC"),
+              *table->schema()->field(0)->type());
+    auto ts =
+        std::static_pointer_cast<::arrow::TimestampArray>(table->column(0)->chunk(0));
+    ASSERT_EQ(2, ts->length());
+    ASSERT_EQ(1000000, ts->Value(0));
+    ASSERT_EQ(INT64_MAX, ts->Value(1));
+  }
+}
+
 TEST(TestArrowReadWrite, ImplicitSecondToMillisecondTimestampCoercion) {
   using ::arrow::ArrayFromVector;
   using ::arrow::field;

cpp/src/parquet/arrow/arrow_schema_test.cc

@@ -263,6 +263,12 @@ TEST_F(TestConvertParquetSchema, ParquetAnnotatedFields) {
        ::arrow::fixed_size_binary(16)},
       {"float16", LogicalType::Float16(), ParquetType::FIXED_LEN_BYTE_ARRAY, 2,
        ::arrow::float16()},
+      {"timestamp_flba12_ms", LogicalType::Timestamp(true, LogicalType::TimeUnit::MILLIS),
+       ParquetType::FIXED_LEN_BYTE_ARRAY, 12, ::arrow::fixed_size_binary(12)},
+      {"timestamp_flba12_us", LogicalType::Timestamp(true, LogicalType::TimeUnit::MICROS),
+       ParquetType::FIXED_LEN_BYTE_ARRAY, 12, ::arrow::fixed_size_binary(12)},
+      {"timestamp_flba12_ns", LogicalType::Timestamp(true, LogicalType::TimeUnit::NANOS),
+       ParquetType::FIXED_LEN_BYTE_ARRAY, 12, ::arrow::fixed_size_binary(12)},
       {"none", LogicalType::None(), ParquetType::BOOLEAN, -1, ::arrow::boolean()},
       {"none", LogicalType::None(), ParquetType::INT32, -1, ::arrow::int32()},
       {"none", LogicalType::None(), ParquetType::INT64, -1, ::arrow::int64()},
@@ -306,6 +312,29 @@ TEST_F(TestConvertParquetSchema, DuplicateFieldNames) {
   ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(::arrow::schema(arrow_fields)));
 }
 
+TEST_F(TestConvertParquetSchema, FlbaTimestampConversion) {
+  auto make_fields = [] {
+    std::vector<NodePtr> fields;
+    fields.push_back(
+        PrimitiveNode::Make("ts", Repetition::REQUIRED,
+                            LogicalType::Timestamp(true, LogicalType::TimeUnit::MICROS),
+                            ParquetType::FIXED_LEN_BYTE_ARRAY, /*length=*/12));
+    return fields;
+  };
+
+  // Should output the raw FLBA value.
+  ASSERT_OK(ConvertSchema(make_fields()));
+  ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(
+      ::arrow::schema({::arrow::field("ts", ::arrow::fixed_size_binary(12), false)})));
+
+  // Should convert to an Arrow timestamp.
+  ArrowReaderProperties props;
+  props.set_convert_flba_timestamps(true);
+  ASSERT_OK(ConvertSchema(make_fields(), /*key_value_metadata=*/{}, props));
+  ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(::arrow::schema({::arrow::field(
+      "ts", ::arrow::timestamp(::arrow::TimeUnit::MICRO, "UTC"), false)})));
+}
+
 TEST_F(TestConvertParquetSchema, ParquetKeyValueMetadata) {
   std::vector<NodePtr> parquet_fields;
   std::vector<std::shared_ptr<Field>> arrow_fields;

cpp/src/parquet/arrow/reader_internal.cc

@@ -28,6 +28,7 @@
 #include <vector>
 
 #include "arrow/array.h"
+#include "arrow/builder.h"
 #include "arrow/compute/api.h"
 #include "arrow/datum.h"
 #include "arrow/io/memory.h"
@@ -855,6 +856,49 @@ Status TransferHalfFloat(RecordReader* reader, MemoryPool* pool,
   return Status::OK();
 }
 
+// Read a TIMESTAMP-annotated FLBA(12) column as a 64-bit Arrow timestamp. Values that do
+// not fit in the 64 bit range either error or clamp to min/max int64, depending on
+// configuration.
+Status TransferFlbaTimestamp(RecordReader* reader, MemoryPool* pool,
+                             const std::shared_ptr<Field>& field, Datum* out,
+                             bool clamp_on_overflow) {
+  static const auto binary_type = ::arrow::fixed_size_binary(12);
+  std::shared_ptr<ChunkedArray> chunked_array;
+  RETURN_NOT_OK(
+      TransferBinary(reader, pool, field->WithType(binary_type), &chunked_array));
+
+  ::arrow::TimestampBuilder builder(field->type(), pool);
+  RETURN_NOT_OK(builder.Reserve(chunked_array->length()));
+  for (const auto& chunk : chunked_array->chunks()) {
+    const auto& values = checked_cast<const ::arrow::FixedSizeBinaryArray&>(*chunk);
+    for (int64_t i = 0; i < values.length(); ++i) {
+      if (values.IsNull(i)) {
+        builder.UnsafeAppendNull();
+        continue;
+      }
+      const uint8_t* bytes = values.GetValue(i);
+      const uint64_t low = bit_util::FromLittleEndian(SafeLoadAs<uint64_t>(bytes));
+      const uint32_t high = bit_util::FromLittleEndian(SafeLoadAs<uint32_t>(bytes + 8));
+      const int64_t low_signed = static_cast<int64_t>(low);
+      // Fits in int64 iff the high part is a pure sign-extension of the low part.
+      if (static_cast<int32_t>(high) != (low_signed < 0 ? -1 : 0)) {
+        if (!clamp_on_overflow) {
+          return Status::Invalid(
+              "FLBA(12) TIMESTAMP value does not fit in a 64-bit Arrow timestamp");
+        }
+        const bool negative = (bytes[11] & 0x80) != 0;
+        builder.UnsafeAppend(negative ? INT64_MIN : INT64_MAX);
+      } else {
+        builder.UnsafeAppend(low_signed);
+      }
+    }
+  }
+  std::shared_ptr<::arrow::Array> array;
+  RETURN_NOT_OK(builder.Finish(&array));
+  *out = array;
+  return Status::OK();
+}
+
 }  // namespace
 
 #define TRANSFER_INT32(ENUM, ArrowType)                                            \
@@ -966,6 +1010,10 @@ Status TransferColumnData(RecordReader* reader,
       if (descr->physical_type() == ::parquet::Type::INT96) {
         RETURN_NOT_OK(
             TransferInt96(reader, pool, value_field, &result, timestamp_type.unit()));
+      } else if (descr->physical_type() == ::parquet::Type::FIXED_LEN_BYTE_ARRAY) {
+        RETURN_NOT_OK(TransferFlbaTimestamp(
+            reader, pool, value_field, &result,
+            ctx->reader_properties->flba_timestamp_clamp_on_overflow()));
       } else {
         switch (timestamp_type.unit()) {
           case ::arrow::TimeUnit::MILLI:

cpp/src/parquet/arrow/schema_internal.cc

@@ -207,6 +207,13 @@ Result<std::shared_ptr<ArrowType>> FromFLBA(
         return ::arrow::extension::uuid();
       }
 
+      return ::arrow::fixed_size_binary(physical_length);
+    case LogicalType::Type::TIMESTAMP:
+      // If configured, convert to a potentially lossy Arrow timestamp. Otherwise, return
+      // the raw lossless FLBA value.
+      if (physical_length == 12 && reader_properties.convert_flba_timestamps()) {
+        return MakeArrowTimestamp(logical_type);
+      }
       return ::arrow::fixed_size_binary(physical_length);
     default:
       return Status::NotImplemented("Unhandled logical_type ", logical_type.ToString(),

cpp/src/parquet/properties.h

@@ -1157,7 +1157,9 @@ class PARQUET_EXPORT ArrowReaderProperties {
         list_type_(kArrowDefaultListType),
         arrow_extensions_enabled_(false),
         should_load_statistics_(false),
-        smallest_decimal_enabled_(false) {}
+        smallest_decimal_enabled_(false),
+        convert_flba_timestamps_(false),
+        flba_timestamp_clamp_on_overflow_(false) {}
 
   /// \brief Set whether to use the IO thread pool to parse columns in parallel.
   ///
@@ -1295,6 +1297,29 @@ class PARQUET_EXPORT ArrowReaderProperties {
   /// this setting will be ignored.
   bool smallest_decimal_enabled() const { return smallest_decimal_enabled_; }
 
+  /// \brief Set whether to infer Arrow timestamps from Parquet FLBA types.
+  ///
+  /// When enabled, Parquet FLBA(12) TIMESTAMP columns are read as Arrow timestamps.
+  /// VAlues that do not fit in 64 bit timestamps are handled per
+  /// flba_timestamp_clamp_on_overflow(). When disabled, Parquet FLBA(12) TIMESTAMP
+  /// columns are read as FixedSizeBinary(12).
+  void set_convert_flba_timestamps(bool convert) { convert_flba_timestamps_ = convert; }
+  /// \brief Whether FLBA(12) TIMESTAMP columns are read as Arrow timestamps.
+  bool convert_flba_timestamps() const { return convert_flba_timestamps_; }
+
+  /// \brief Set how out-of-range values are handled when convert_flba_timestamps() is
+  /// enabled.
+  ///
+  /// When true, Parquet FLBA(12) TIMESTAMP values that do not fit in 64 bit timestamps
+  /// are clamped to min/max INT64. When false, such values raise an error.
+  void set_flba_timestamp_clamp_on_overflow(bool clamp) {
+    flba_timestamp_clamp_on_overflow_ = clamp;
+  }
+  /// \brief Whether out-of-range FLBA(12) timestamps clamp (true) or error (false).
+  bool flba_timestamp_clamp_on_overflow() const {
+    return flba_timestamp_clamp_on_overflow_;
+  }
+
  private:
   bool use_threads_;
   std::unordered_set<int> read_dict_indices_;
@@ -1308,6 +1333,8 @@ class PARQUET_EXPORT ArrowReaderProperties {
   bool arrow_extensions_enabled_;
   bool should_load_statistics_;
   bool smallest_decimal_enabled_;
+  bool convert_flba_timestamps_;
+  bool flba_timestamp_clamp_on_overflow_;
 };
 
 /// EXPERIMENTAL: Constructs the default ArrowReaderProperties

cpp/src/parquet/reader_test.cc

@@ -138,6 +138,8 @@ std::string byte_stream_split_extended() {
   return data_file("byte_stream_split_extended.gzip.parquet");
 }
 
+std::string flba12_timestamp() { return data_file("flba12_timestamp.parquet"); }
+
 template <typename DType, typename ValueType = typename DType::c_type>
 std::vector<ValueType> ReadColumnValues(ParquetFileReader* file_reader, int row_group,
                                         int column, int64_t expected_values_read) {
@@ -1769,6 +1771,73 @@ TEST(TestByteStreamSplit, ExtendedIntegrationFile) {
 }
 #endif  // ARROW_WITH_ZLIB
 
+TEST(TestFileReader, TestFlba12Timestamp) {
+  auto file = ParquetFileReader::OpenFile(flba12_timestamp());
+
+  const int64_t kNumRows = 6;
+  // Row indices of the minimum (year 0001) and maximum (year 9999) values.
+  const int kMinRow = 5;
+  const int kMaxRow = 4;
+
+  auto metadata = file->metadata();
+  ASSERT_EQ(kNumRows, metadata->num_rows());
+  ASSERT_EQ(3, metadata->num_columns());
+  ASSERT_EQ(1, metadata->num_row_groups());
+
+  const struct {
+    const char* name;
+    LogicalType::TimeUnit::unit unit;
+  } columns[] = {
+      {"timestamp_millis", LogicalType::TimeUnit::MILLIS},
+      {"timestamp_micros", LogicalType::TimeUnit::MICROS},
+      {"timestamp_nanos", LogicalType::TimeUnit::NANOS},
+  };
+
+  auto rg_reader = file->RowGroup(0);
+  for (int c = 0; c < 3; ++c) {
+    const auto* descr = metadata->schema()->Column(c);
+    ASSERT_EQ(columns[c].name, descr->name());
+    ASSERT_EQ(Type::FIXED_LEN_BYTE_ARRAY, descr->physical_type());
+    ASSERT_EQ(12, descr->type_length());
+    ASSERT_EQ(SortOrder::SIGNED, descr->sort_order());
+    ASSERT_EQ(ColumnOrder::TYPE_DEFINED_ORDER, descr->column_order().get_order());
+
+    const auto& logical_type = descr->logical_type();
+    ASSERT_EQ(LogicalType::Type::TIMESTAMP, logical_type->type());
+    const auto& ts =
+        ::arrow::internal::checked_cast<const TimestampLogicalType&>(*logical_type);
+    ASSERT_TRUE(ts.is_adjusted_to_utc());
+    ASSERT_EQ(columns[c].unit, ts.time_unit());
+
+    std::string min_value, max_value;
+    {
+      auto col_reader =
+          checked_pointer_cast<TypedColumnReader<FLBAType>>(rg_reader->Column(c));
+      std::vector<FLBA> values(kNumRows);
+      int64_t values_read = 0;
+      int64_t levels_read =
+          col_reader->ReadBatch(kNumRows, nullptr, nullptr, values.data(), &values_read);
+      ASSERT_EQ(kNumRows, levels_read);
+      ASSERT_EQ(kNumRows, values_read);
+      min_value.assign(reinterpret_cast<const char*>(values[kMinRow].ptr), 12);
+      max_value.assign(reinterpret_cast<const char*>(values[kMaxRow].ptr), 12);
+
+      auto comparator = MakeComparator<FLBAType>(descr);
+      auto min_max = comparator->GetMinMax(values.data(), kNumRows);
+      ASSERT_EQ(min_value,
+                std::string(reinterpret_cast<const char*>(min_max.first.ptr), 12));
+      ASSERT_EQ(max_value,
+                std::string(reinterpret_cast<const char*>(min_max.second.ptr), 12));
+    }
+
+    auto stats = rg_reader->metadata()->ColumnChunk(c)->statistics();
+    ASSERT_NE(nullptr, stats);
+    ASSERT_TRUE(stats->HasMinMax());
+    ASSERT_EQ(min_value, stats->EncodeMin());
+    ASSERT_EQ(max_value, stats->EncodeMax());
+  }
+}
+
 struct PageIndexReaderParam {
   std::vector<int32_t> row_group_indices;
   std::vector<int32_t> column_indices;

cpp/src/parquet/schema_test.cc

@@ -1417,6 +1417,12 @@ TEST(TestLogicalTypeOperation, LogicalTypeApplicability) {
   for (const InapplicableType& t : inapplicable_types) {
     ASSERT_FALSE(logical_type->is_applicable(t.physical_type, t.physical_length));
   }
+
+  // TIMESTAMP is applicable to INT64 and FLBA(12).
+  logical_type = LogicalType::Timestamp(true, LogicalType::TimeUnit::MILLIS);
+  ASSERT_TRUE(logical_type->is_applicable(Type::INT64));
+  ASSERT_TRUE(logical_type->is_applicable(Type::FIXED_LEN_BYTE_ARRAY, 12));
+  ASSERT_FALSE(logical_type->is_applicable(Type::FIXED_LEN_BYTE_ARRAY, 8));
 }
 
 TEST(TestLogicalTypeOperation, DecimalLogicalTypeApplicability) {