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,80 @@ 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 and -1,000,000 (both fit int64),
+  // 2^64 (overflows INT64_MAX), and -2^64 (underflows INT64_MIN).
+  uint8_t pos_in_range[12] = {0x40, 0x42, 0x0f, 0, 0, 0, 0, 0, 0, 0, 0, 0};
+  uint8_t neg_in_range[12] = {0xc0, 0xbd, 0xf0, 0xff, 0xff, 0xff,
+                              0xff, 0xff, 0xff, 0xff, 0xff, 0xff};
+  uint8_t overflow[12] = {0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0};
+  uint8_t neg_overflow[12] = {0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 0xff, 0xff};
+  FLBA values[4] = {FLBA(pos_in_range), FLBA(neg_in_range), FLBA(overflow),
+                    FLBA(neg_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(4, 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));
+    ARROW_ASSIGN_OR_RAISE(*out, reader->ReadTable());
+    return ::arrow::Status::OK();
+  };
+
+  // Convert, error on overflow (default): the out-of-range rows fail the read.
+  {
+    ArrowReaderProperties props;
+    std::shared_ptr<Table> table;
+    ASSERT_RAISES(Invalid, read_table(props, &table));
+  }
+
+  // Conversion disabled: raw, lossless FixedSizeBinary(12).
+  {
+    ArrowReaderProperties props;
+    props.set_convert_flba_timestamps(false);
+    std::shared_ptr<Table> table;
+    ASSERT_OK(read_table(props, &table));
+    ASSERT_EQ(::arrow::Type::FIXED_SIZE_BINARY, table->schema()->field(0)->type()->id());
+  }
+
+  // Convert, clamp on overflow: in-range value is exact; positive overflow clamps
+  // to INT64_MAX and negative overflow clamps to INT64_MIN.
+  {
+    ArrowReaderProperties props;
+    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(4, ts->length());
+    ASSERT_EQ(1000000, ts->Value(0));
+    ASSERT_EQ(-1000000, ts->Value(1));
+    ASSERT_EQ(INT64_MAX, ts->Value(2));
+    ASSERT_EQ(INT64_MIN, ts->Value(3));
+  }
+}
+
 TEST(TestArrowReadWrite, ImplicitSecondToMillisecondTimestampCoercion) {
   using ::arrow::ArrayFromVector;
   using ::arrow::field;

cpp/src/parquet/arrow/arrow_schema_test.cc

@@ -263,6 +263,15 @@ 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::timestamp(::arrow::TimeUnit::MILLI, "UTC")},
+      {"timestamp_flba12_us", LogicalType::Timestamp(true, LogicalType::TimeUnit::MICROS),
+       ParquetType::FIXED_LEN_BYTE_ARRAY, 12,
+       ::arrow::timestamp(::arrow::TimeUnit::MICRO, "UTC")},
+      {"timestamp_flba12_ns", LogicalType::Timestamp(true, LogicalType::TimeUnit::NANOS),
+       ParquetType::FIXED_LEN_BYTE_ARRAY, 12,
+       ::arrow::timestamp(::arrow::TimeUnit::NANO, "UTC")},
       {"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 +315,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 convert to an Arrow timestamp.
+  ASSERT_OK(ConvertSchema(make_fields()));
+  ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(::arrow::schema({::arrow::field(
+      "ts", ::arrow::timestamp(::arrow::TimeUnit::MICRO, "UTC"), false)})));
+
+  // Should output the raw FLBA value.
+  ArrowReaderProperties props;
+  props.set_convert_flba_timestamps(false);
+  ASSERT_OK(ConvertSchema(make_fields(), /*key_value_metadata=*/{}, props));
+  ASSERT_NO_FATAL_FAILURE(CheckFlatSchema(
+      ::arrow::schema({::arrow::field("ts", ::arrow::fixed_size_binary(12), 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

@@ -46,6 +46,7 @@
 #include "arrow/util/int_util_overflow.h"
 #include "arrow/util/logging_internal.h"
 #include "arrow/util/ubsan.h"
+#include "arrow/visit_data_inline.h"
 
 #include "parquet/arrow/reader.h"
 #include "parquet/arrow/schema.h"
@@ -855,6 +856,79 @@ Status TransferHalfFloat(RecordReader* reader, MemoryPool* pool,
   return Status::OK();
 }
 
+// Decode a little-endian 96-bit FLBA(12) TIMESTAMP value into a 64-bit Arrow timestamp.
+// Values that do not fit in the int64 range either error or clamp to INT64_MIN/INT64_MAX,
+// depending on clamp_on_overflow.
+Status FlbaTimestampToInt64(const uint8_t* bytes, bool clamp_on_overflow, int64_t* out) {
+  const uint64_t low = bit_util::FromLittleEndian(SafeLoadAs<uint64_t>(bytes));
+  const uint32_t high = bit_util::FromLittleEndian(SafeLoadAs<uint32_t>(bytes + 8));
+  const int32_t high_signed = static_cast<int32_t>(high);
+  const int64_t low_signed = static_cast<int64_t>(low);
+  const int32_t sign_extension = (low_signed < 0) ? -1 : 0;
+  // Fits in int64 iff the high part is a pure sign-extension of the low part.
+  if (high_signed != sign_extension) {
+    if (!clamp_on_overflow) {
+      return Status::Invalid(
+          "FLBA(12) TIMESTAMP value does not fit in a 64-bit Arrow timestamp");
+    }
+    *out = high_signed < 0 ? INT64_MIN : INT64_MAX;
+  } else {
+    *out = low_signed;
+  }
+  return Status::OK();
+}
+
+Result<::arrow::TimeUnit::type> ArrowTimeUnitFromParquet(LogicalType::TimeUnit::unit unit) {
+  switch (unit) {
+    case LogicalType::TimeUnit::MILLIS:
+      return ::arrow::TimeUnit::MILLI;
+    case LogicalType::TimeUnit::MICROS:
+      return ::arrow::TimeUnit::MICRO;
+    case LogicalType::TimeUnit::NANOS:
+      return ::arrow::TimeUnit::NANO;
+    default:
+      return Status::Invalid("Unrecognized Parquet TIMESTAMP time unit");
+  }
+}
+
+// Read a TIMESTAMP-annotated FLBA(12) column as a 64-bit Arrow timestamp.
+Status TransferFlbaTimestamp(RecordReader* reader, MemoryPool* pool,
+                             const std::shared_ptr<Field>& field, Datum* out,
+                             bool clamp_on_overflow) {
+  auto binary_reader = dynamic_cast<BinaryRecordReader*>(reader);
+  DCHECK(binary_reader);
+  ::arrow::ArrayVector chunks = binary_reader->GetBuilderChunks();
+
+  for (size_t i = 0; i < chunks.size(); ++i) {
+    const auto& values = checked_cast<const ::arrow::FixedSizeBinaryArray&>(*chunks[i]);
+    const int64_t length = values.length();
+    ARROW_ASSIGN_OR_RAISE(auto data,
+                          ::arrow::AllocateBuffer(length * sizeof(int64_t), pool));
+    auto out_ptr = reinterpret_cast<int64_t*>(data->mutable_data());
+
+    int64_t j = 0;
+    RETURN_NOT_OK(::arrow::VisitArraySpanInline<::arrow::FixedSizeBinaryType>(
+        ::arrow::ArraySpan(*values.data()),
+        [&](std::string_view v) {
+          return FlbaTimestampToInt64(reinterpret_cast<const uint8_t*>(v.data()),
+                                      clamp_on_overflow, &out_ptr[j++]);
+        },
+        [&]() {
+          out_ptr[j++] = 0;
+          return ::arrow::Status::OK();
+        }));
+
+    chunks[i] = std::make_shared<::arrow::TimestampArray>(
+        field->type(), length, std::move(data), values.null_bitmap(),
+        values.null_count());
+  }
+  if (!field->nullable()) {
+    ReconstructChunksWithoutNulls(&chunks);
+  }
+  *out = std::make_shared<ChunkedArray>(std::move(chunks), field->type());
+  return Status::OK();
+}
+
 }  // namespace
 
 #define TRANSFER_INT32(ENUM, ArrowType)                                            \
@@ -966,6 +1040,21 @@ 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) {
+        // Validate that the provided Arrow timestamp unit matches the Parquet unit.
+        const auto& ts_logical =
+            checked_cast<const TimestampLogicalType&>(*descr->logical_type());
+        ARROW_ASSIGN_OR_RAISE(auto expected_unit,
+                              ArrowTimeUnitFromParquet(ts_logical.time_unit()));
+        if (timestamp_type.unit() != expected_unit) {
+          return Status::Invalid(
+              "Arrow timestamp unit ", timestamp_type.unit(),
+              " does not match Parquet FLBA(12) TIMESTAMP logical type ",
+              ts_logical.ToString());
+        }
+        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,12 @@ 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.
+      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_(true),
+        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) {