pitrou commented on code in PR #50916:
URL: https://github.com/apache/arrow/pull/50916#discussion_r4120889547
##########
cpp/src/parquet/arrow/reader_internal.cc:
##########
@@ -853,6 +854,80 @@ 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) {
Review Comment:
Mark this unlikely as most timestamps will not be that large, presumably?
```suggestion
if (ARROW_PREDICT_FALSE(high_signed != sign_extension)) {
```
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2166,6 +2166,80 @@ TEST(TestArrowReadWrite, CoerceTimestampsLosePrecision) {
allow_truncation_to_micros));
}
Review Comment:
Can you add an integration test against "flba12_timestamp.parquet"?
##########
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 {
Review Comment:
Let's please return `Result<std::shared_ptr<Table>>`.
##########
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());
Review Comment:
Please let's validate the table: call `Table::ValidateFull` on the result.
##########
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));
Review Comment:
Can you assert something about the error message? Something like:
```suggestion
EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid,
::testing::HasSubstr("something bad happened"),
read_table(props, &table));
```
##########
cpp/src/parquet/arrow/reader_internal.cc:
##########
@@ -853,6 +854,80 @@ 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) {
Review Comment:
Can you return `Result<int64_t>` instead of the out-parameter?
##########
cpp/src/parquet/arrow/reader_internal.cc:
##########
@@ -853,6 +854,80 @@ 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;
Review Comment:
Can we use `std::numeric_limits` instead of the old C constants?
##########
cpp/src/parquet/arrow/reader_internal.cc:
##########
@@ -853,6 +854,80 @@ 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) {
Review Comment:
We probably have similar code elsewhere already, why not reuse it?
##########
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());
Review Comment:
Please check the table contents, not just lazily its type.
##########
cpp/src/parquet/arrow/reader_internal.cc:
##########
@@ -853,6 +854,80 @@ 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) {
Review Comment:
Also, should probably make this function `inline`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
