This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8f2b222455c [fix](be) Correct pre-epoch Parquet timestamp conversion
(#67834)
8f2b222455c is described below
commit 8f2b222455ca47fb28a554539f7b0f932bc5ad8d
Author: Oliveira <[email protected]>
AuthorDate: Sun Sep 13 20:22:59 2026 +0800
[fix](be) Correct pre-epoch Parquet timestamp conversion (#67834)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Loading Parquet timestamps before the Unix epoch with
fractional seconds could produce a negative subsecond component. Its
conversion to an unsigned value then corrupted the timestamp, yielding
NULL or a 1970 value. Normalize the quotient and remainder with floor
semantics before constructing the timestamp.
### Release note
Fix loading pre-epoch Parquet timestamps that contain fractional
seconds.
### Check List (For Author)
- Test: Unit Test / Regression test
- Ran 4 focused Parquet timestamp conversion BE unit tests
- Ran test_outfile_parquet regression suite against a local FE/BE
cluster, including exact values for five pre-epoch timestamps
- Behavior changed: Yes (pre-epoch Parquet timestamps with fractional
seconds are decoded correctly)
- Does this need documentation: No
---
be/src/format/parquet/parquet_column_convert.h | 61 ++++++----
.../format/parquet/parquet_column_convert_test.cpp | 129 ++++++++++++++++++++-
.../outfile/parquet/test_outfile_parquet.out | 6 +
.../outfile/parquet/test_outfile_parquet.groovy | 66 +++++++++++
4 files changed, 240 insertions(+), 22 deletions(-)
diff --git a/be/src/format/parquet/parquet_column_convert.h
b/be/src/format/parquet/parquet_column_convert.h
index 5ec05d450d8..e91018b932a 100644
--- a/be/src/format/parquet/parquet_column_convert.h
+++ b/be/src/format/parquet/parquet_column_convert.h
@@ -43,6 +43,25 @@
namespace doris::parquet {
namespace detail {
+struct EpochSecondsAndMicros {
+ int64_t seconds;
+ uint32_t microseconds;
+};
+
+inline EpochSecondsAndMicros split_epoch_time(int64_t timestamp, int64_t
units_per_second,
+ int64_t nanos_per_unit) {
+ int64_t seconds = timestamp / units_per_second;
+ int64_t subsecond = timestamp % units_per_second;
+ // C++ division truncates toward zero. Normalize to floor division so the
fractional part is
+ // always non-negative for timestamps before the Unix epoch.
+ if (subsecond < 0) {
+ subsecond += units_per_second;
+ --seconds;
+ }
+ return {.seconds = seconds,
+ .microseconds = static_cast<uint32_t>(subsecond * nanos_per_unit /
1000)};
+}
+
inline bool try_split_local_time(int64_t local_time, uint16_t* year, uint8_t*
month, uint8_t* day,
uint8_t* hour, uint8_t* minute, uint8_t*
second) {
static const libdivide::divider<int64_t> fast_div_86400(86400);
@@ -833,18 +852,18 @@ struct Int64ToTimestamp : public
PhysicalToLogicalConverter {
int64_t x = src_data[i];
auto& num = data[start_idx + i];
auto& value =
reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(num);
- const int64_t epoch_seconds = x / _convert_params->second_mask;
+ const auto epoch_time = detail::split_epoch_time(x,
_convert_params->second_mask,
+
_convert_params->scale_to_nano_factor);
if (_convert_params->is_fixed_offset) {
if (!detail::try_convert_timestamp_with_fixed_offset(
- value, epoch_seconds,
_convert_params->fixed_offset_seconds)) {
- value.from_unixtime(epoch_seconds, *_convert_params->ctz);
+ value, epoch_time.seconds,
_convert_params->fixed_offset_seconds)) {
+ value.from_unixtime(epoch_time.seconds,
*_convert_params->ctz);
}
- } else if (!detail::try_convert_timestamp_with_lookup(value,
epoch_seconds,
+ } else if (!detail::try_convert_timestamp_with_lookup(value,
epoch_time.seconds,
*_convert_params->ctz)) {
- value.from_unixtime(epoch_seconds, *_convert_params->ctz);
+ value.from_unixtime(epoch_time.seconds, *_convert_params->ctz);
}
- value.set_microsecond((x % _convert_params->second_mask) *
- (_convert_params->scale_to_nano_factor /
1000));
+ value.set_microsecond(epoch_time.microseconds);
}
return Status::OK();
}
@@ -866,9 +885,10 @@ struct Int64ToTimestampTz : public
PhysicalToLogicalConverter {
for (int i = 0; i < rows; i++) {
int64_t x = src_data[i];
auto& tz = dest_data[start_idx + i];
- tz.from_unixtime(x / _convert_params->second_mask, UTC);
- tz.set_microsecond((x % _convert_params->second_mask) *
- (_convert_params->scale_to_nano_factor / 1000));
+ const auto epoch_time = detail::split_epoch_time(x,
_convert_params->second_mask,
+
_convert_params->scale_to_nano_factor);
+ tz.from_unixtime(epoch_time.seconds, UTC);
+ tz.set_microsecond(epoch_time.microseconds);
}
return Status::OK();
}
@@ -891,18 +911,18 @@ struct Int96toTimestamp : public
PhysicalToLogicalConverter {
auto& dst_value =
reinterpret_cast<DateV2Value<DateTimeV2ValueType>&>(data[start_idx + i]);
- int64_t timestamp_with_micros =
src_cell_data.to_timestamp_micros();
- const int64_t epoch_seconds = timestamp_with_micros / 1000000;
+ const auto epoch_time =
+
detail::split_epoch_time(src_cell_data.to_timestamp_micros(), 1000000, 1000);
if (_convert_params->is_fixed_offset) {
if (!detail::try_convert_timestamp_with_fixed_offset(
- dst_value, epoch_seconds,
_convert_params->fixed_offset_seconds)) {
- dst_value.from_unixtime(epoch_seconds,
*_convert_params->ctz);
+ dst_value, epoch_time.seconds,
_convert_params->fixed_offset_seconds)) {
+ dst_value.from_unixtime(epoch_time.seconds,
*_convert_params->ctz);
}
- } else if (!detail::try_convert_timestamp_with_lookup(dst_value,
epoch_seconds,
+ } else if (!detail::try_convert_timestamp_with_lookup(dst_value,
epoch_time.seconds,
*_convert_params->ctz)) {
- dst_value.from_unixtime(epoch_seconds, *_convert_params->ctz);
+ dst_value.from_unixtime(epoch_time.seconds,
*_convert_params->ctz);
}
- dst_value.set_microsecond(timestamp_with_micros % 1000000);
+ dst_value.set_microsecond(epoch_time.microseconds);
}
return Status::OK();
}
@@ -923,10 +943,11 @@ struct Int96toTimestampTz : public
PhysicalToLogicalConverter {
for (int i = 0; i < rows; i++) {
ParquetInt96 src_cell_data = ParquetInt96_data[i];
- int64_t timestamp_with_micros =
src_cell_data.to_timestamp_micros();
auto& tz = data[start_idx + i];
- tz.from_unixtime(timestamp_with_micros / 1000000, UTC);
- tz.set_microsecond(timestamp_with_micros % 1000000);
+ const auto epoch_time =
+
detail::split_epoch_time(src_cell_data.to_timestamp_micros(), 1000000, 1000);
+ tz.from_unixtime(epoch_time.seconds, UTC);
+ tz.set_microsecond(epoch_time.microseconds);
}
return Status::OK();
}
diff --git a/be/test/format/parquet/parquet_column_convert_test.cpp
b/be/test/format/parquet/parquet_column_convert_test.cpp
index 112390442dc..1f12a090030 100644
--- a/be/test/format/parquet/parquet_column_convert_test.cpp
+++ b/be/test/format/parquet/parquet_column_convert_test.cpp
@@ -20,6 +20,8 @@
#include <cctz/time_zone.h>
#include <chrono>
+#include <cstring>
+#include <string>
#include <vector>
#include "core/assert_cast.h"
@@ -30,17 +32,60 @@
namespace doris::parquet {
-static FieldSchema make_timestamp_field_schema(bool is_adjusted_to_utc) {
+enum class TestTimestampUnit { MILLIS, MICROS, NANOS };
+
+static FieldSchema make_timestamp_field_schema(bool is_adjusted_to_utc,
+ TestTimestampUnit unit =
TestTimestampUnit::MICROS) {
FieldSchema field_schema;
field_schema.parquet_schema.__set_name("ts");
field_schema.parquet_schema.__set_logicalType(tparquet::LogicalType());
field_schema.parquet_schema.logicalType.__set_TIMESTAMP(tparquet::TimestampType());
field_schema.parquet_schema.logicalType.TIMESTAMP.__set_isAdjustedToUTC(is_adjusted_to_utc);
field_schema.parquet_schema.logicalType.TIMESTAMP.__set_unit(tparquet::TimeUnit());
-
field_schema.parquet_schema.logicalType.TIMESTAMP.unit.__set_MICROS(tparquet::MicroSeconds());
+ auto& parquet_unit =
field_schema.parquet_schema.logicalType.TIMESTAMP.unit;
+ switch (unit) {
+ case TestTimestampUnit::MILLIS:
+ parquet_unit.__set_MILLIS(tparquet::MilliSeconds());
+ break;
+ case TestTimestampUnit::MICROS:
+ parquet_unit.__set_MICROS(tparquet::MicroSeconds());
+ break;
+ case TestTimestampUnit::NANOS:
+ parquet_unit.__set_NANOS(tparquet::NanoSeconds());
+ break;
+ }
return field_schema;
}
+static void expect_int64_timestamp(int64_t timestamp, TestTimestampUnit unit,
+ const std::string& expected, bool
is_adjusted_to_utc = false,
+ const cctz::time_zone* timezone = nullptr) {
+ auto field_schema = make_timestamp_field_schema(is_adjusted_to_utc, unit);
+ field_schema.parquet_schema.__set_type(tparquet::Type::INT64);
+ field_schema.data_type =
+ DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2,
false, 0, 6);
+
+ auto converter = PhysicalToLogicalConverter::get_converter(
+ &field_schema, field_schema.data_type, field_schema.data_type,
timezone);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+
+ auto src_column = ColumnInt64::create();
+ src_column->insert_value(timestamp);
+ ColumnPtr src = std::move(src_column);
+ ColumnPtr dst = field_schema.data_type->create_column();
+ ASSERT_TRUE(converter->physical_convert(src, dst).ok());
+ ASSERT_EQ(dst->size(), 1);
+ EXPECT_EQ(field_schema.data_type->to_string(*dst, 0), expected);
+}
+
+static ColumnPtr make_int96_column(const ParquetInt96& timestamp) {
+ auto column = ColumnInt8::create();
+ auto& data = column->get_data();
+ data.resize(sizeof(timestamp));
+ std::memcpy(data.data(), ×tamp, sizeof(timestamp));
+ return column;
+}
+
TEST(ParquetColumnConvertTest, InitFixedOffsetDetection) {
TimezoneUtils::load_timezones_to_cache();
@@ -125,6 +170,86 @@ TEST(ParquetColumnConvertTest, LookupPathMatchesOriginal) {
}
}
+TEST(ParquetColumnConvertTest, ConvertsInt64TimestampsAcrossEpoch) {
+ struct TestCase {
+ int64_t timestamp;
+ TestTimestampUnit unit;
+ const char* expected;
+ };
+ const std::vector<TestCase> test_cases {
+ {-1, TestTimestampUnit::MILLIS, "1969-12-31 23:59:59.999000"},
+ {-500, TestTimestampUnit::MILLIS, "1969-12-31 23:59:59.500000"},
+ {-1000, TestTimestampUnit::MILLIS, "1969-12-31 23:59:59.000000"},
+ {-1500, TestTimestampUnit::MILLIS, "1969-12-31 23:59:58.500000"},
+ {-1, TestTimestampUnit::MICROS, "1969-12-31 23:59:59.999999"},
+ {-500000, TestTimestampUnit::MICROS, "1969-12-31 23:59:59.500000"},
+ {-500000000, TestTimestampUnit::NANOS, "1969-12-31
23:59:59.500000"},
+ {0, TestTimestampUnit::MICROS, "1970-01-01 00:00:00.000000"},
+ {500000123, TestTimestampUnit::NANOS, "1970-01-01
00:00:00.500000"},
+ };
+
+ for (const auto& test_case : test_cases) {
+ SCOPED_TRACE(testing::Message()
+ << "timestamp=" << test_case.timestamp << ", expected="
<< test_case.expected);
+ expect_int64_timestamp(test_case.timestamp, test_case.unit,
test_case.expected);
+ }
+}
+
+TEST(ParquetColumnConvertTest, ConvertsNegativeInt64TimestampWithTimezone) {
+ const auto plus_eight = cctz::fixed_time_zone(std::chrono::hours(8));
+ expect_int64_timestamp(-500000, TestTimestampUnit::MICROS, "1970-01-01
07:59:59.500000", true,
+ &plus_eight);
+}
+
+TEST(ParquetColumnConvertTest, ConvertsNegativeInt96TimestampsBeforeEpoch) {
+ auto field_schema = make_timestamp_field_schema(false);
+ field_schema.parquet_schema.__set_type(tparquet::Type::INT96);
+ const ParquetInt96 timestamp {
+ .lo = 86399500000000,
+ .hi = ParquetInt96::JULIAN_EPOCH_OFFSET_DAYS - 1,
+ };
+
+ for (const auto primitive_type : {TYPE_DATETIMEV2, TYPE_TIMESTAMPTZ}) {
+ field_schema.data_type =
+ DataTypeFactory::instance().create_data_type(primitive_type,
false, 0, 6);
+ auto converter = PhysicalToLogicalConverter::get_converter(
+ &field_schema, field_schema.data_type, field_schema.data_type,
nullptr);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+
+ ColumnPtr src = make_int96_column(timestamp);
+ ColumnPtr dst = field_schema.data_type->create_column();
+ ASSERT_TRUE(converter->physical_convert(src, dst).ok());
+ ASSERT_EQ(dst->size(), 1);
+
+ if (primitive_type == TYPE_DATETIMEV2) {
+ EXPECT_EQ(field_schema.data_type->to_string(*dst, 0), "1969-12-31
23:59:59.500000");
+ } else {
+ const auto& column = assert_cast<const ColumnTimeStampTz&>(*dst);
+ EXPECT_EQ(column.get_data()[0].utc_dt().to_string(6), "1969-12-31
23:59:59.500000");
+ }
+ }
+}
+
+TEST(ParquetColumnConvertTest, ConvertsNegativeInt64TimestampTzBeforeEpoch) {
+ auto field_schema = make_timestamp_field_schema(true);
+ field_schema.parquet_schema.__set_type(tparquet::Type::INT64);
+ field_schema.data_type =
+ DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ,
false, 0, 6);
+ auto converter = PhysicalToLogicalConverter::get_converter(
+ &field_schema, field_schema.data_type, field_schema.data_type,
nullptr);
+ ASSERT_TRUE(converter->support()) << converter->get_error_msg();
+
+ auto src_column = ColumnInt64::create();
+ src_column->insert_value(-500000);
+ ColumnPtr src = std::move(src_column);
+ ColumnPtr dst = field_schema.data_type->create_column();
+ ASSERT_TRUE(converter->physical_convert(src, dst).ok());
+ ASSERT_EQ(dst->size(), 1);
+
+ const auto& column = assert_cast<const ColumnTimeStampTz&>(*dst);
+ EXPECT_EQ(column.get_data()[0].utc_dt().to_string(6), "1969-12-31
23:59:59.500000");
+}
+
TEST(ParquetColumnConvertTest, AlignNullMapUsesAppendedSourceSlice) {
auto dst_nested_column = ColumnFloat64::create();
dst_nested_column->insert_value(1);
diff --git
a/regression-test/data/export_p0/outfile/parquet/test_outfile_parquet.out
b/regression-test/data/export_p0/outfile/parquet/test_outfile_parquet.out
index cb6eab32684..eab1032420b 100644
--- a/regression-test/data/export_p0/outfile/parquet/test_outfile_parquet.out
+++ b/regression-test/data/export_p0/outfile/parquet/test_outfile_parquet.out
@@ -23,3 +23,9 @@
9 2017-10-01 2017-10-01T00:00 2017-10-01
2017-10-01T00:00 2017-10-01T00:00:00.111 2017-10-01T00:00:00.111111
Beijing 9 9 true 9 9 9 9.9 9.9 char9
9
10 2017-10-01 2017-10-01T00:00 2017-10-01
2017-10-01T00:00 2017-10-01T00:00:00.111 2017-10-01T00:00:00.111111
\N \N \N \N \N \N \N \N \N \N
\N
+-- !negative_timestamp --
+1 1900-01-01T00:00:00.123456
+2 1969-12-31T23:59:58.500
+3 1969-12-31T23:59:59.500
+4 1969-12-31T23:59:59.999
+5 1969-12-31T23:59:59.999999
diff --git
a/regression-test/suites/export_p0/outfile/parquet/test_outfile_parquet.groovy
b/regression-test/suites/export_p0/outfile/parquet/test_outfile_parquet.groovy
index f87e656c7c1..5600b01be32 100644
---
a/regression-test/suites/export_p0/outfile/parquet/test_outfile_parquet.groovy
+++
b/regression-test/suites/export_p0/outfile/parquet/test_outfile_parquet.groovy
@@ -66,6 +66,7 @@ suite("test_outfile_parquet") {
def tableName2 = "outfile_parquet_test2"
def uuid = UUID.randomUUID().toString()
def outFilePath = """/tmp/test_outfile_parquet_${uuid}"""
+ def negativeTimestampOutFilePath =
"""/tmp/test_outfile_parquet_negative_timestamp_${uuid}"""
try {
sql """ DROP TABLE IF EXISTS ${tableName} """
sql """
@@ -158,6 +159,64 @@ suite("test_outfile_parquet") {
logger.info("Run command: command=" + command + ",code=" + code + ",
out=" + out + ", err=" + err)
assertEquals(code, 0)
qt_select_default """ SELECT * FROM ${tableName2} t ORDER BY user_id;
"""
+
+ sql """ DROP TABLE IF EXISTS outfile_parquet_negative_timestamp """
+ sql """
+ CREATE TABLE outfile_parquet_negative_timestamp (
+ `id` INT NOT NULL,
+ `ts` DATETIMEV2(6) NOT NULL
+ )
+ DUPLICATE KEY (`id`, `ts`)
+ PARTITION BY RANGE (`ts`) (
+ PARTITION p_pre_epoch VALUES LESS THAN ("1970-01-01 00:00:00")
+ )
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES("replication_num" = "1")
+ """
+
+ File negativeTimestampPath = new File(negativeTimestampOutFilePath)
+ assert negativeTimestampPath.mkdirs()
+ sql """
+ SELECT id, ts
+ FROM (
+ SELECT 1 AS id, CAST('1900-01-01 00:00:00.123456' AS
DATETIMEV2(6)) AS ts
+ UNION ALL
+ SELECT 2 AS id, CAST('1969-12-31 23:59:58.500000' AS
DATETIMEV2(6)) AS ts
+ UNION ALL
+ SELECT 3 AS id, CAST('1969-12-31 23:59:59.500000' AS
DATETIMEV2(6)) AS ts
+ UNION ALL
+ SELECT 4 AS id, CAST('1969-12-31 23:59:59.999000' AS
DATETIMEV2(6)) AS ts
+ UNION ALL
+ SELECT 5 AS id, CAST('1969-12-31 23:59:59.999999' AS
DATETIMEV2(6)) AS ts
+ ) t
+ ORDER BY id
+ INTO OUTFILE "file://${negativeTimestampOutFilePath}/"
+ FORMAT AS PARQUET
+ PROPERTIES("enable_int96_timestamps" = "false")
+ """
+
+ File[] negativeTimestampFiles = negativeTimestampPath.listFiles()
+ assert negativeTimestampFiles.length == 1
+ commandBuilder = new StringBuilder()
+ commandBuilder.append("""curl -v --location-trusted -u
${context.config.feHttpUser}:${context.config.feHttpPassword}""")
+ commandBuilder.append(""" -H format:parquet -T """ +
negativeTimestampFiles[0].getAbsolutePath() + """
http://${context.config.feHttpAddress}/api/"""; + dbName +
"/outfile_parquet_negative_timestamp/_stream_load")
+ command = commandBuilder.toString()
+ process = command.execute()
+ code = process.waitFor()
+ err = IOGroovyMethods.getText(new BufferedReader(new
InputStreamReader(process.getErrorStream())))
+ out = process.getText()
+ logger.info("Run negative timestamp stream load: command=" + command +
",code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def loadResponse = parseJson(out.trim())
+ assertEquals("Success", loadResponse.Status)
+ assertEquals(5, loadResponse.NumberTotalRows)
+ assertEquals(5, loadResponse.NumberLoadedRows)
+ assertEquals(0, loadResponse.NumberFilteredRows)
+ order_qt_negative_timestamp """
+ SELECT id, ts
+ FROM outfile_parquet_negative_timestamp
+ ORDER BY id
+ """
} finally {
try_sql("DROP TABLE IF EXISTS ${tableName}")
try_sql("DROP TABLE IF EXISTS ${tableName2}")
@@ -168,5 +227,12 @@ suite("test_outfile_parquet") {
}
path.delete();
}
+ File negativeTimestampPath = new File(negativeTimestampOutFilePath)
+ if (negativeTimestampPath.exists()) {
+ for (File f: negativeTimestampPath.listFiles()) {
+ f.delete();
+ }
+ negativeTimestampPath.delete();
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
