Skip to content

Commit b9c5839

Browse files
Gabriel39Gabriel39
committed
[fix](external) Separate Hudi JNI timestamp timezones
Keep logical INT64 Hudi timestamps on the query session timezone while applying hive.parquet.time-zone only to INT96 in scanner v2. Preserve scanner v1 legacy behavior.
1 parent 26804c2 commit b9c5839

6 files changed

Lines changed: 90 additions & 27 deletions

File tree

‎be/src/format/table/hudi_jni_reader.cpp‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -56,13 +56,7 @@ HudiJniReader::HudiJniReader(const TFileScanRangeParams& scan_params,
5656
{"instant_time", hudi_params.instant_time},
5757
{"serde", hudi_params.serde},
5858
{"input_format", hudi_params.input_format},
59-
// An unset catalog zone means raw INT96 wall time, represented by UTC
60-
// in JNI, rather than inheriting an unrelated query session zone.
61-
{"time_zone",
62-
scan_params.__isset.hive_parquet_time_zone &&
63-
!scan_params.hive_parquet_time_zone.empty()
64-
? scan_params.hive_parquet_time_zone
65-
: "UTC"}};
59+
{"time_zone", state->timezone_obj().name()}};
6660
for (const auto& kv : scan_params.properties) {
6761
if (kv.first.starts_with(HOODIE_CONF_PREFIX)) {
6862
params[kv.first] = kv.second;

‎be/src/format_v2/jni/hudi_jni_reader.cpp‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -93,10 +93,12 @@ Status HudiJniReader::build_scanner_params(std::map<std::string, std::string>* p
9393
(*params)["instant_time"] = hudi_params.instant_time;
9494
(*params)["serde"] = hudi_params.serde;
9595
(*params)["input_format"] = hudi_params.input_format;
96-
(*params)["time_zone"] = _scan_params->__isset.hive_parquet_time_zone &&
97-
!_scan_params->hive_parquet_time_zone.empty()
98-
? _scan_params->hive_parquet_time_zone
99-
: "UTC";
96+
// Keep the INT96 compatibility zone separate because Hudi JNI also uses the session zone to
97+
// materialize logical INT64 timestamps, which must match native Parquet splits.
98+
(*params)["int96_time_zone"] = _scan_params->__isset.hive_parquet_time_zone &&
99+
!_scan_params->hive_parquet_time_zone.empty()
100+
? _scan_params->hive_parquet_time_zone
101+
: "UTC";
100102
if (_runtime_state != nullptr) {
101103
(*params)["query_id"] = print_id(_runtime_state->query_id());
102104
}

‎be/test/format_v2/jni/jni_table_reader_test.cpp‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,7 @@ TEST(JniTableReaderTest, GenericConnectorUsesQuerySessionTimezone) {
165165
EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
166166
}
167167

168-
TEST(HudiJniReaderTest, CatalogInt96TimezoneOverridesQuerySession) {
168+
TEST(HudiJniReaderTest, CatalogInt96TimezoneIsSeparateFromQuerySession) {
169169
TFileScanRangeParams scan_params;
170170
scan_params.__set_hive_parquet_time_zone("Asia/Shanghai");
171171
RuntimeState state {TQueryOptions(), TQueryGlobals()};
@@ -191,7 +191,8 @@ TEST(HudiJniReaderTest, CatalogInt96TimezoneOverridesQuerySession) {
191191
ASSERT_TRUE(reader.build_scanner_params(&params).ok());
192192
reader._scanner_params = std::move(params);
193193
reader._apply_common_scanner_params();
194-
EXPECT_EQ(reader._scanner_params["time_zone"], "Asia/Shanghai");
194+
EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
195+
EXPECT_EQ(reader._scanner_params["int96_time_zone"], "Asia/Shanghai");
195196
}
196197

197198
TEST(HudiJniReaderTest, UnconfiguredInt96TimezoneUsesRawWallClockPolicy) {
@@ -219,7 +220,8 @@ TEST(HudiJniReaderTest, UnconfiguredInt96TimezoneUsesRawWallClockPolicy) {
219220
ASSERT_TRUE(reader.build_scanner_params(&params).ok());
220221
reader._scanner_params = std::move(params);
221222
reader._apply_common_scanner_params();
222-
EXPECT_EQ(reader._scanner_params["time_zone"], "UTC");
223+
EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
224+
EXPECT_EQ(reader._scanner_params["int96_time_zone"], "UTC");
223225
}
224226

225227
TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) {

‎fe/be-java-extensions/hadoop-hudi-scanner/src/main/java/org/apache/doris/hudi/HadoopHudiColumnValue.java‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -46,10 +46,12 @@ public class HadoopHudiColumnValue implements ColumnValue {
4646
private ColumnType dorisType;
4747
private ObjectInspector fieldInspector;
4848
private Object fieldData;
49-
private final ZoneId zoneId;
49+
private final ZoneId sessionZoneId;
50+
private final ZoneId int96ZoneId;
5051

51-
public HadoopHudiColumnValue(ZoneId zoneId) {
52-
this.zoneId = zoneId;
52+
public HadoopHudiColumnValue(ZoneId sessionZoneId, ZoneId int96ZoneId) {
53+
this.sessionZoneId = sessionZoneId;
54+
this.int96ZoneId = int96ZoneId;
5355
}
5456

5557
public void setRow(Object record) {
@@ -133,7 +135,7 @@ public LocalDateTime getDateTime() {
133135
return ((Timestamp) fieldData).toLocalDateTime();
134136
} else if (fieldData instanceof TimestampWritableV2) {
135137
return LocalDateTime.ofInstant(Instant.ofEpochSecond((((TimestampObjectInspector) fieldInspector)
136-
.getPrimitiveJavaObject(fieldData)).toEpochSecond()), zoneId);
138+
.getPrimitiveJavaObject(fieldData)).toEpochSecond()), int96ZoneId);
137139
} else {
138140
long datetime = ((LongWritable) fieldData).get();
139141
long seconds;
@@ -148,7 +150,9 @@ public LocalDateTime getDateTime() {
148150
throw new RuntimeException("Hoodie timestamp only support milliseconds and microseconds, "
149151
+ "wrong precision = " + dorisType.getPrecision());
150152
}
151-
return LocalDateTime.ofInstant(Instant.ofEpochSecond(seconds, nanoseconds), zoneId);
153+
// Logical INT64 timestamps carry epoch values and must remain in the session zone;
154+
// applying the INT96 override would make native and JNI Hudi splits disagree.
155+
return LocalDateTime.ofInstant(Instant.ofEpochSecond(seconds, nanoseconds), sessionZoneId);
152156
}
153157
}
154158

@@ -179,7 +183,7 @@ public void unpackArray(List<ColumnValue> values) {
179183
ObjectInspector itemInspector = inspector.getListElementObjectInspector();
180184
for (int i = 0; i < items.size(); i++) {
181185
Object item = items.get(i);
182-
HadoopHudiColumnValue childValue = new HadoopHudiColumnValue(zoneId);
186+
HadoopHudiColumnValue childValue = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
183187
childValue.setRow(item);
184188
childValue.setField(dorisType.getChildTypes().get(0), itemInspector);
185189
values.add(childValue);
@@ -192,12 +196,12 @@ public void unpackMap(List<ColumnValue> keys, List<ColumnValue> values) {
192196
ObjectInspector keyObjectInspector = inspector.getMapKeyObjectInspector();
193197
ObjectInspector valueObjectInspector = inspector.getMapValueObjectInspector();
194198
for (Map.Entry kv : inspector.getMap(fieldData).entrySet()) {
195-
HadoopHudiColumnValue key = new HadoopHudiColumnValue(zoneId);
199+
HadoopHudiColumnValue key = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
196200
key.setRow(kv.getKey());
197201
key.setField(dorisType.getChildTypes().get(0), keyObjectInspector);
198202
keys.add(key);
199203

200-
HadoopHudiColumnValue value = new HadoopHudiColumnValue(zoneId);
204+
HadoopHudiColumnValue value = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
201205
value.setRow(kv.getValue());
202206
value.setField(dorisType.getChildTypes().get(1), valueObjectInspector);
203207
values.add(value);
@@ -210,7 +214,7 @@ public void unpackStruct(List<Integer> structFieldIndex, List<ColumnValue> value
210214
List<? extends StructField> fields = inspector.getAllStructFieldRefs();
211215
for (int i = 0; i < structFieldIndex.size(); i++) {
212216
Integer idx = structFieldIndex.get(i);
213-
HadoopHudiColumnValue value = new HadoopHudiColumnValue(zoneId);
217+
HadoopHudiColumnValue value = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
214218
Object obj = null;
215219
if (idx != null) {
216220
StructField sf = fields.get(idx);

‎fe/be-java-extensions/hadoop-hudi-scanner/src/main/java/org/apache/doris/hudi/HadoopHudiJniScanner.java‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -131,13 +131,21 @@ public HadoopHudiJniScanner(int fetchSize, Map<String, String> params) {
131131
}
132132
this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(fsOptionsProps);
133133

134-
ZoneId zoneId;
134+
ZoneId sessionZoneId;
135135
if (Strings.isNullOrEmpty(params.get("time_zone"))) {
136-
zoneId = ZoneId.systemDefault();
136+
sessionZoneId = ZoneId.systemDefault();
137137
} else {
138-
zoneId = ZoneId.of(params.get("time_zone"));
138+
sessionZoneId = ZoneId.of(params.get("time_zone"));
139139
}
140-
this.columnValue = new HadoopHudiColumnValue(zoneId);
140+
ZoneId int96ZoneId;
141+
if (Strings.isNullOrEmpty(params.get("int96_time_zone"))) {
142+
// Scanner v1 does not send the v2-only override and must keep its legacy session-zone
143+
// behavior for every timestamp representation.
144+
int96ZoneId = sessionZoneId;
145+
} else {
146+
int96ZoneId = ZoneId.of(params.get("int96_time_zone"));
147+
}
148+
this.columnValue = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
141149
this.fetchSize = fetchSize;
142150
this.classLoader = this.getClass().getClassLoader();
143151
}

‎fe/be-java-extensions/hadoop-hudi-scanner/src/test/java/org/apache/doris/hudi/HadoopHudiColumnValueTest.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.hudi;
19+
20+
import org.apache.doris.common.jni.vec.ColumnType;
21+
22+
import org.apache.hadoop.hive.common.type.Timestamp;
23+
import org.apache.hadoop.hive.serde2.io.TimestampWritableV2;
24+
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
25+
import org.apache.hadoop.io.LongWritable;
26+
import org.junit.Assert;
27+
import org.junit.Test;
28+
29+
import java.time.LocalDateTime;
30+
import java.time.ZoneId;
31+
32+
public class HadoopHudiColumnValueTest {
33+
@Test
34+
public void testInt64TimestampUsesSessionTimezone() {
35+
HadoopHudiColumnValue value = new HadoopHudiColumnValue(
36+
ZoneId.of("America/Los_Angeles"), ZoneId.of("Asia/Shanghai"));
37+
value.setField(ColumnType.parseType("ts", "datetimev2(6)"), null);
38+
value.setRow(new LongWritable(0));
39+
40+
Assert.assertEquals(LocalDateTime.of(1969, 12, 31, 16, 0), value.getDateTime());
41+
}
42+
43+
@Test
44+
public void testInt96TimestampUsesCompatibilityTimezone() {
45+
HadoopHudiColumnValue value = new HadoopHudiColumnValue(
46+
ZoneId.of("America/Los_Angeles"), ZoneId.of("Asia/Shanghai"));
47+
value.setField(ColumnType.parseType("ts", "datetimev2(6)"),
48+
PrimitiveObjectInspectorFactory.writableTimestampObjectInspector);
49+
value.setRow(new TimestampWritableV2(Timestamp.ofEpochSecond(0)));
50+
51+
Assert.assertEquals(LocalDateTime.of(1970, 1, 1, 8, 0), value.getDateTime());
52+
}
53+
}

0 commit comments

Comments
 (0)