Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
divjotarora wants to merge 2 commits into
base: main
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0f1904f
Allow TIMESTAMP logical type to annotate FIXED_LEN_BYTE_ARRAY(12)
divjotarora Aug 19, 2026
97b3d4e
address comments
divjotarora Aug 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions cpp/src/parquet/arrow/arrow_schema_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()},
Expand Down
2 changes: 2 additions & 0 deletions cpp/src/parquet/arrow/schema_internal.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,8 @@ Result<std::shared_ptr<ArrowType>> FromFLBA(
return ::arrow::extension::uuid();
}

return ::arrow::fixed_size_binary(physical_length);
case LogicalType::Type::TIMESTAMP:
return ::arrow::fixed_size_binary(physical_length);

@emkornfield emkornfield Aug 20, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this should probably be configurable. We should probably have a mode that takes returns the arrow timestamp type (and either errors on overflow or converts to MIN/MAX representable values., maybe a different config value?)

@divjotarora divjotarora Aug 20, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ack, added convert_flba_timestamps and flba_timestamp_clamp_on_overflow properties to control conversion from FLBA(12) --> Arrow timestamps and clamping to min/max int64 vs. erroring for values out of the int64 range, respectively

default:
return Status::NotImplemented("Unhandled logical_type ", logical_type.ToString(),
Expand Down
62 changes: 62 additions & 0 deletions cpp/src/parquet/reader_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -1769,6 +1771,66 @@ 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);
Comment thread
divjotarora marked this conversation as resolved.
}

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;
Expand Down
6 changes: 6 additions & 0 deletions cpp/src/parquet/schema_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
54 changes: 54 additions & 0 deletions cpp/src/parquet/statistics.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,56 @@ struct RebindLogical<Float16LogicalType> {
using c_type = DType::c_type;
};

// Tag type for FLBA(12) timestamps.
struct Flba12TimestampType {};

// Max / min representable signed 96-bit two's-complement, little-endian.
constexpr uint8_t kFlba12SignedMax[12] = {0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F};
constexpr uint8_t kFlba12SignedMin[12] = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x80};

template <>
struct CompareHelper<Flba12TimestampType, /*is_signed=*/true> {
using T = FLBA;

// Seed for the running minimum is the maximum value; for the maximum, the minimum
// value.
static T DefaultMin() { return T{kFlba12SignedMax}; }
static T DefaultMax() { return T{kFlba12SignedMin}; }

static T Coalesce(T val, T fallback) { return val.ptr == nullptr ? fallback : val; }

// Signed little-endian comparison.
// Differing signs: negative (MSB >= 0x80) is smaller.
// Same sign: unsigned scan and comparison.
static inline bool Compare(int /*type_length*/, const T& a, const T& b) {
const bool a_neg = (a.ptr[11] & 0x80) != 0;
const bool b_neg = (b.ptr[11] & 0x80) != 0;
if (a_neg != b_neg) return a_neg;
for (int i = 11; i >= 0; --i) if (a.ptr[i] != b.ptr[i]) return a.ptr[i] < b.ptr[i];
Comment thread
divjotarora marked this conversation as resolved.
Outdated
return false;
}

static T Min(int type_length, const T& a, const T& b) {
if (a.ptr == nullptr) return b;
if (b.ptr == nullptr) return a;
return Compare(type_length, a, b) ? a : b;
}

static T Max(int type_length, const T& a, const T& b) {
if (a.ptr == nullptr) return b;
if (b.ptr == nullptr) return a;
return Compare(type_length, a, b) ? b : a;
}
};

template <>
struct RebindLogical<Flba12TimestampType> {
using DType = FLBAType;
using c_type = DType::c_type;
};

template <bool is_signed, typename DType>
class TypedComparatorImpl
: virtual public TypedComparator<typename RebindLogical<DType>::DType> {
Expand Down Expand Up @@ -1014,6 +1064,10 @@ std::shared_ptr<Comparator> DoMakeComparator(Type::type physical_type,
return std::make_shared<TypedComparatorImpl<true, Float16LogicalType>>(
type_length);
}
if (logical_type == LogicalType::Type::TIMESTAMP) {
return std::make_shared<TypedComparatorImpl<true, Flba12TimestampType>>(
type_length);
}
return std::make_shared<TypedComparatorImpl<true, FLBAType>>(type_length);
default:
ParquetException::NYI("Signed Compare not implemented");
Expand Down
67 changes: 67 additions & 0 deletions cpp/src/parquet/statistics_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,6 +175,73 @@ TEST(Comparison, SignedFLBA) {
}
}

TEST(Comparison, SignedLittleEndianFLBA12Timestamp) {
NodePtr node =
PrimitiveNode::Make("ts_flba12", Repetition::REQUIRED,
LogicalType::Timestamp(true, LogicalType::TimeUnit::NANOS),
Type::FIXED_LEN_BYTE_ARRAY, 12);
ColumnDescriptor descr(node, 0, 0);
ASSERT_EQ(SortOrder::SIGNED, descr.sort_order());
auto comparator = MakeComparator<FLBAType>(&descr);

// large_neg: 0x80 00…00 (most negative 96-bit LE value, sign byte=0x80)
// neg256: 0x00 FF FF…FF (= -256 in LE two's complement)
// minus_one: FF FF…FF (= -1)
// zero: 00 00…00
// plus_one: 01 00…00
// large_pos: 7F FF…FF (= near max positive)
std::vector<uint8_t> large_neg_bytes = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x80};
std::vector<uint8_t> neg256_bytes = {0x00, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF};
std::vector<uint8_t> minus_one_bytes = {0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF};
std::vector<uint8_t> zero_bytes = {0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
std::vector<uint8_t> plus_one_bytes = {0x01, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
std::vector<uint8_t> large_pos_bytes = {0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F};

std::vector<FLBA> vals = {FLBA(large_neg_bytes.data()), FLBA(neg256_bytes.data()),
FLBA(minus_one_bytes.data()), FLBA(zero_bytes.data()),
FLBA(plus_one_bytes.data()), FLBA(large_pos_bytes.data())};

for (size_t x = 0; x < vals.size(); x++) {
EXPECT_FALSE(comparator->Compare(vals[x], vals[x])) << x;
for (size_t y = x + 1; y < vals.size(); y++) {
EXPECT_TRUE(comparator->Compare(vals[x], vals[y])) << x << " < " << y;
EXPECT_FALSE(comparator->Compare(vals[y], vals[x])) << y << " < " << x;
}
}
}

TEST(Comparison, SignedLittleEndianFLBA12TimestampMinMax) {
// Guards the DefaultMin/DefaultMax accumulator seeds: over positive-only values.
NodePtr node =
PrimitiveNode::Make("ts_flba12", Repetition::REQUIRED,
LogicalType::Timestamp(true, LogicalType::TimeUnit::NANOS),
Type::FIXED_LEN_BYTE_ARRAY, 12);
ColumnDescriptor descr(node, 0, 0);
auto comparator = MakeComparator<FLBAType>(&descr);

std::vector<uint8_t> plus_one = {0x01, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
std::vector<uint8_t> plus_five = {0x05, 0x00, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, 0x00, 0x00, 0x00, 0x00};
std::vector<uint8_t> large_pos = {0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F};
std::vector<FLBA> vals = {FLBA(plus_five.data()), FLBA(large_pos.data()),
FLBA(plus_one.data())};

auto min_max = comparator->GetMinMax(vals.data(), vals.size());
// min == plus_one and max == large_pos.
EXPECT_FALSE(comparator->Compare(min_max.first, FLBA(plus_one.data())));
EXPECT_FALSE(comparator->Compare(FLBA(plus_one.data()), min_max.first));
EXPECT_FALSE(comparator->Compare(min_max.second, FLBA(large_pos.data())));
EXPECT_FALSE(comparator->Compare(FLBA(large_pos.data()), min_max.second));
}

TEST(Comparison, UnsignedFLBA) {
int size = 10;
auto comparator =
Expand Down
11 changes: 9 additions & 2 deletions cpp/src/parquet/types.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1376,10 +1376,12 @@ LogicalType::TimeUnit::unit TimeLogicalType::time_unit() const {
}

class LogicalType::Impl::Timestamp final : public LogicalType::Impl::Compatible,
public LogicalType::Impl::SimpleApplicable {
public LogicalType::Impl::Applicable {
public:
friend class TimestampLogicalType;

bool is_applicable(parquet::Type::type primitive_type,
int32_t primitive_length = -1) const override;
bool is_serialized() const override;
bool is_compatible(ConvertedType::type converted_type,
schema::DecimalMetadata converted_decimal_metadata) const override;
Expand All @@ -1400,7 +1402,6 @@ class LogicalType::Impl::Timestamp final : public LogicalType::Impl::Compatible,
Timestamp(bool adjusted, LogicalType::TimeUnit::unit unit, bool is_from_converted_type,
bool force_set_converted_type)
: LogicalType::Impl(LogicalType::Type::TIMESTAMP, SortOrder::SIGNED),
LogicalType::Impl::SimpleApplicable(parquet::Type::INT64),
adjusted_(adjusted),
unit_(unit),
is_from_converted_type_(is_from_converted_type),
Expand All @@ -1411,6 +1412,12 @@ class LogicalType::Impl::Timestamp final : public LogicalType::Impl::Compatible,
bool force_set_converted_type_ = false;
};

bool LogicalType::Impl::Timestamp::is_applicable(parquet::Type::type primitive_type,
int32_t primitive_length) const {
return primitive_type == parquet::Type::INT64 ||
(primitive_type == parquet::Type::FIXED_LEN_BYTE_ARRAY && primitive_length == 12);
}

bool LogicalType::Impl::Timestamp::is_serialized() const {
return !is_from_converted_type_;
}
Expand Down