[fix](external) Preserve Hudi and Paimon timestamp semantics

Issue Number: None

Related PR: #65446

Problem Summary:

Hudi scanner v2 started applying the Hive INT96 compatibility timezone, which changed Hudi existing session-timezone behavior and could make native and JNI splits disagree. Paimon nested filter-only TIMESTAMP_LTZ columns could also lose their table-format timestamp annotation while building the file projection. In addition, a fuzzy session variable could disable COUNT pushdown for a multi-gigabyte MAP regression fixture and cause scanner v2 to materialize the payload under ASAN.

Keep Hudi native and JNI timestamp decoding on the SQL session timezone, propagate Paimon timestamp semantics into filter-only projections, and cover the precision-9 INT96 path. Make COUNT pushdown deterministic only for the large MAP fixture so scanner v2 reads definition levels while scanner v1 retains its existing behavior.

Author: Gabriel39

be/src/format_v2/column_mapper_nested.cpp

@@ -545,6 +545,9 @@ Status build_file_child_projection_from_schema(const std::vector<ColumnDefinitio
         return Status::OK();
     }
     *projection = LocalColumnIndex::local(child->file_local_id());
+    // Filter-only paths may have no ColumnMapping, so their schema projection must carry the
+    // table-format timestamp semantic itself.
+    projection->timestamp_is_adjusted_to_utc = child->timestamp_is_adjusted_to_utc;
     projection->project_all_children = selectors.size() == 1;
     projection->children.clear();
     if (selectors.size() == 1) {

be/src/format_v2/jni/hudi_jni_reader.cpp

@@ -93,12 +93,6 @@ Status HudiJniReader::build_scanner_params(std::map<std::string, std::string>* p
     (*params)["instant_time"] = hudi_params.instant_time;
     (*params)["serde"] = hudi_params.serde;
     (*params)["input_format"] = hudi_params.input_format;
-    // Keep the INT96 compatibility zone separate because Hudi JNI also uses the session zone to
-    // materialize logical INT64 timestamps, which must match native Parquet splits.
-    (*params)["int96_time_zone"] = _scan_params->__isset.hive_parquet_time_zone &&
-                                                   !_scan_params->hive_parquet_time_zone.empty()
-                                           ? _scan_params->hive_parquet_time_zone
-                                           : "UTC";
     if (_runtime_state != nullptr) {
         (*params)["query_id"] = print_id(_runtime_state->query_id());
     }

be/src/format_v2/table/hudi_reader.cpp

@@ -22,6 +22,7 @@
 #include "exprs/vexpr_context.h"
 #include "format_v2/column_mapper.h"
 #include "format_v2/jni/hudi_jni_reader.h"
+#include "format_v2/parquet/parquet_reader.h"
 #include "format_v2/table/schema_history_util.h"
 #include "gen_cpp/PlanNodes_types.h"
 
@@ -60,6 +61,31 @@ format::TableColumnMappingMode HudiReader::mapping_mode() const {
                    : format::TableColumnMappingMode::BY_NAME;
 }
 
+Status HudiReader::create_file_reader(std::unique_ptr<format::FileReader>* reader) {
+    if (_format != format::FileFormat::PARQUET) {
+        return format::TableReader::create_file_reader(reader);
+    }
+    DORIS_CHECK(reader != nullptr);
+    const bool enable_mapping_timestamp_tz = _scan_params != nullptr &&
+                                             _scan_params->__isset.enable_mapping_timestamp_tz &&
+                                             _scan_params->enable_mapping_timestamp_tz;
+    const bool enable_mapping_varbinary = _scan_params != nullptr &&
+                                          _scan_params->__isset.enable_mapping_varbinary &&
+                                          _scan_params->enable_mapping_varbinary;
+    *reader = std::make_unique<format::parquet::ParquetReader>(
+            _system_properties, _current_task->data_file, _io_ctx, _scanner_profile,
+            _global_rowid_context, enable_mapping_timestamp_tz, enable_mapping_varbinary,
+            parquet_int96_time_zone());
+    return Status::OK();
+}
+
+std::string HudiReader::parquet_int96_time_zone() const {
+    DORIS_CHECK(_runtime_state != nullptr);
+    // Hudi keeps its legacy session-timezone contract for native base files; Hive's INT96
+    // compatibility property must not make native and JNI Hudi splits diverge.
+    return _runtime_state->timezone();
+}
+
 Status HudiReader::annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) {
     DORIS_CHECK(file_schema != nullptr);
     if (mapping_mode() != format::TableColumnMappingMode::BY_FIELD_ID) {

be/src/format_v2/table/hudi_reader.h

@@ -36,13 +36,16 @@ class HudiReader final : public format::TableReader {
 #ifdef BE_TEST
     void TEST_set_scan_params(TFileScanRangeParams* params) { _scan_params = params; }
     format::TableColumnMappingMode TEST_mapping_mode() const { return mapping_mode(); }
+    std::string TEST_parquet_int96_time_zone() const { return parquet_int96_time_zone(); }
     Status TEST_annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) {
         return annotate_file_schema(file_schema);
     }
 #endif
 
 protected:
+    Status create_file_reader(std::unique_ptr<format::FileReader>* reader) override;
     format::TableColumnMappingMode mapping_mode() const override;
+    std::string parquet_int96_time_zone() const;
     Status annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) override;
 
 private:

be/test/format_v2/column_mapper_test.cpp

@@ -3062,6 +3062,43 @@ TEST(ColumnMapperScanRequestTest, PredicateProjectionRebuildsProjectedStructFile
     EXPECT_FALSE(mapper.mappings()[0].is_trivial);
 }
 
+// Scenario: Paimon projects one struct child but filters on an unprojected TIMESTAMP_LTZ(9)
+// child. The filter-only file projection must retain the history-schema timestamp semantic so an
+// unannotated INT96 leaf is materialized as TIMESTAMPTZ instead of DATETIMEV2.
+TEST(ColumnMapperScanRequestTest, FilterOnlyNestedTimestampRetainsTableFormatSemantic) {
+    const auto int_type = i32();
+    const auto ltz_type = timestamptz(9);
+
+    auto table_payload = field_id_col("payload", 1, int_type);
+    auto projected_table_struct = struct_col("s", 10, {table_payload});
+    auto table_ltz = field_id_col("ltz", 2, ltz_type);
+    auto full_table_struct = struct_col("s", 10, {table_payload, table_ltz});
+
+    auto file_payload = field_id_col("payload", 1, int_type, 0);
+    auto file_ltz = field_id_col("ltz", 2, ltz_type, 1);
+    file_ltz.timestamp_is_adjusted_to_utc = true;
+    auto file_struct = struct_col("s", 10, {file_payload, file_ltz}, 5);
+
+    TableColumnMapper mapper({.mode = TableColumnMappingMode::BY_FIELD_ID});
+    ASSERT_TRUE(mapper.create_mapping({projected_table_struct}, {}, {file_struct}).ok());
+
+    auto filter_expr = null_predicate(
+            struct_element(table_slot(0, 0, full_table_struct.type, "s"), ltz_type, "ltz"), false);
+    TableFilter filter {.conjunct = VExprContext::create_shared(filter_expr),
+                        .global_indices = {GlobalIndex(0)}};
+
+    FileScanRequest request;
+    ASSERT_TRUE(mapper.create_scan_request({filter}, {projected_table_struct}, &request).ok());
+
+    ASSERT_EQ(request.predicate_columns.size(), 1);
+    const auto& root_projection = request.predicate_columns[0];
+    ASSERT_EQ(projection_ids(root_projection.children), std::vector<int32_t>({0, 1}));
+    const auto* ltz_projection = find_child_projection(&root_projection, 1);
+    ASSERT_NE(ltz_projection, nullptr);
+    ASSERT_TRUE(ltz_projection->timestamp_is_adjusted_to_utc.has_value());
+    EXPECT_TRUE(*ltz_projection->timestamp_is_adjusted_to_utc);
+}
+
 // Scenario: a filter references a top-level column that is not projected by the query; the mapper
 // creates a hidden filter mapping without adding that hidden column to visible table mappings.
 TEST(ColumnMapperScanRequestTest, PredicateOnlyTopLevelColumnUsesHiddenMapping) {

be/test/format_v2/jni/jni_table_reader_test.cpp

@@ -165,7 +165,7 @@ TEST(JniTableReaderTest, GenericConnectorUsesQuerySessionTimezone) {
     EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
 }
 
-TEST(HudiJniReaderTest, CatalogInt96TimezoneIsSeparateFromQuerySession) {
+TEST(HudiJniReaderTest, CatalogInt96TimezoneDoesNotOverrideSessionTimezone) {
     TFileScanRangeParams scan_params;
     scan_params.__set_hive_parquet_time_zone("Asia/Shanghai");
     RuntimeState state {TQueryOptions(), TQueryGlobals()};
@@ -192,10 +192,10 @@ TEST(HudiJniReaderTest, CatalogInt96TimezoneIsSeparateFromQuerySession) {
     reader._scanner_params = std::move(params);
     reader._apply_common_scanner_params();
     EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
-    EXPECT_EQ(reader._scanner_params["int96_time_zone"], "Asia/Shanghai");
+    EXPECT_FALSE(reader._scanner_params.contains("int96_time_zone"));
 }
 
-TEST(HudiJniReaderTest, UnconfiguredInt96TimezoneUsesRawWallClockPolicy) {
+TEST(HudiJniReaderTest, UnconfiguredInt96TimezoneUsesSessionTimezone) {
     TFileScanRangeParams scan_params;
     RuntimeState state {TQueryOptions(), TQueryGlobals()};
     state.set_timezone("America/Los_Angeles");
@@ -221,7 +221,7 @@ TEST(HudiJniReaderTest, UnconfiguredInt96TimezoneUsesRawWallClockPolicy) {
     reader._scanner_params = std::move(params);
     reader._apply_common_scanner_params();
     EXPECT_EQ(reader._scanner_params["time_zone"], "America/Los_Angeles");
-    EXPECT_EQ(reader._scanner_params["int96_time_zone"], "UTC");
+    EXPECT_FALSE(reader._scanner_params.contains("int96_time_zone"));
 }
 
 TEST(JniTableReaderTest, CancellationStopsBeforeFetchingAnotherJavaBatch) {

be/test/format_v2/table/hudi_reader_test.cpp

@@ -214,6 +214,25 @@ TEST(HudiReaderTest, FallsBackToByNameWhenSplitHistorySchemaIsMissing) {
     EXPECT_TRUE(file_schema[0].name_mapping.empty());
 }
 
+TEST(HudiReaderTest, NativeInt96KeepsSessionTimezone) {
+    TFileScanRangeParams scan_params;
+    scan_params.__set_hive_parquet_time_zone("Asia/Shanghai");
+    RuntimeState state {TQueryOptions(), TQueryGlobals()};
+    state.set_timezone("America/Los_Angeles");
+
+    hudi::HudiReader reader;
+    ASSERT_TRUE(reader.init({.projected_columns = {},
+                             .conjuncts = {},
+                             .format = FileFormat::PARQUET,
+                             .scan_params = &scan_params,
+                             .io_ctx = nullptr,
+                             .runtime_state = &state,
+                             .scanner_profile = nullptr})
+                        .ok());
+
+    EXPECT_EQ(reader.TEST_parquet_int96_time_zone(), "America/Los_Angeles");
+}
+
 // Scenario: HudiReader must reset the previous split schema id before each split. Otherwise a
 // BY_FIELD_ID split could leak its schema id into the next split that carries no schema id.
 TEST(HudiReaderTest, ResetsSplitSchemaIdBeforePreparingNextSplit) {

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

@@ -46,12 +46,10 @@ public class HadoopHudiColumnValue implements ColumnValue {
     private ColumnType dorisType;
     private ObjectInspector fieldInspector;
     private Object fieldData;
-    private final ZoneId sessionZoneId;
-    private final ZoneId int96ZoneId;
+    private final ZoneId zoneId;
 
-    public HadoopHudiColumnValue(ZoneId sessionZoneId, ZoneId int96ZoneId) {
-        this.sessionZoneId = sessionZoneId;
-        this.int96ZoneId = int96ZoneId;
+    public HadoopHudiColumnValue(ZoneId zoneId) {
+        this.zoneId = zoneId;
     }
 
     public void setRow(Object record) {
@@ -135,7 +133,7 @@ public LocalDateTime getDateTime() {
             return ((Timestamp) fieldData).toLocalDateTime();
         } else if (fieldData instanceof TimestampWritableV2) {
             return LocalDateTime.ofInstant(Instant.ofEpochSecond((((TimestampObjectInspector) fieldInspector)
-                    .getPrimitiveJavaObject(fieldData)).toEpochSecond()), int96ZoneId);
+                    .getPrimitiveJavaObject(fieldData)).toEpochSecond()), zoneId);
         } else {
             long datetime = ((LongWritable) fieldData).get();
             long seconds;
@@ -150,9 +148,7 @@ public LocalDateTime getDateTime() {
                 throw new RuntimeException("Hoodie timestamp only support milliseconds and microseconds, "
                         + "wrong precision = " + dorisType.getPrecision());
             }
-            // Logical INT64 timestamps carry epoch values and must remain in the session zone;
-            // applying the INT96 override would make native and JNI Hudi splits disagree.
-            return LocalDateTime.ofInstant(Instant.ofEpochSecond(seconds, nanoseconds), sessionZoneId);
+            return LocalDateTime.ofInstant(Instant.ofEpochSecond(seconds, nanoseconds), zoneId);
         }
     }
 
@@ -183,7 +179,7 @@ public void unpackArray(List<ColumnValue> values) {
         ObjectInspector itemInspector = inspector.getListElementObjectInspector();
         for (int i = 0; i < items.size(); i++) {
             Object item = items.get(i);
-            HadoopHudiColumnValue childValue = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
+            HadoopHudiColumnValue childValue = new HadoopHudiColumnValue(zoneId);
             childValue.setRow(item);
             childValue.setField(dorisType.getChildTypes().get(0), itemInspector);
             values.add(childValue);
@@ -196,12 +192,12 @@ public void unpackMap(List<ColumnValue> keys, List<ColumnValue> values) {
         ObjectInspector keyObjectInspector = inspector.getMapKeyObjectInspector();
         ObjectInspector valueObjectInspector = inspector.getMapValueObjectInspector();
         for (Map.Entry kv : inspector.getMap(fieldData).entrySet()) {
-            HadoopHudiColumnValue key = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
+            HadoopHudiColumnValue key = new HadoopHudiColumnValue(zoneId);
             key.setRow(kv.getKey());
             key.setField(dorisType.getChildTypes().get(0), keyObjectInspector);
             keys.add(key);
 
-            HadoopHudiColumnValue value = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
+            HadoopHudiColumnValue value = new HadoopHudiColumnValue(zoneId);
             value.setRow(kv.getValue());
             value.setField(dorisType.getChildTypes().get(1), valueObjectInspector);
             values.add(value);
@@ -214,7 +210,7 @@ public void unpackStruct(List<Integer> structFieldIndex, List<ColumnValue> value
         List<? extends StructField> fields = inspector.getAllStructFieldRefs();
         for (int i = 0; i < structFieldIndex.size(); i++) {
             Integer idx = structFieldIndex.get(i);
-            HadoopHudiColumnValue value = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
+            HadoopHudiColumnValue value = new HadoopHudiColumnValue(zoneId);
             Object obj = null;
             if (idx != null) {
                 StructField sf = fields.get(idx);

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

@@ -131,21 +131,14 @@ public HadoopHudiJniScanner(int fetchSize, Map<String, String> params) {
         }
         this.preExecutionAuthenticator = PreExecutionAuthenticatorCache.getAuthenticator(fsOptionsProps);
 
-        ZoneId sessionZoneId;
+        ZoneId zoneId;
         if (Strings.isNullOrEmpty(params.get("time_zone"))) {
-            sessionZoneId = ZoneId.systemDefault();
+            zoneId = ZoneId.systemDefault();
         } else {
-            sessionZoneId = ZoneId.of(params.get("time_zone"));
+            zoneId = ZoneId.of(params.get("time_zone"));
         }
-        ZoneId int96ZoneId;
-        if (Strings.isNullOrEmpty(params.get("int96_time_zone"))) {
-            // Scanner v1 does not send the v2-only override and must keep its legacy session-zone
-            // behavior for every timestamp representation.
-            int96ZoneId = sessionZoneId;
-        } else {
-            int96ZoneId = ZoneId.of(params.get("int96_time_zone"));
-        }
-        this.columnValue = new HadoopHudiColumnValue(sessionZoneId, int96ZoneId);
+        // Hudi keeps one session zone for every timestamp encoding to preserve its JNI contract.
+        this.columnValue = new HadoopHudiColumnValue(zoneId);
         this.fetchSize = fetchSize;
         this.classLoader = this.getClass().getClassLoader();
     }

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

@@ -32,22 +32,20 @@
 public class HadoopHudiColumnValueTest {
     @Test
     public void testInt64TimestampUsesSessionTimezone() {
-        HadoopHudiColumnValue value = new HadoopHudiColumnValue(
-                ZoneId.of("America/Los_Angeles"), ZoneId.of("Asia/Shanghai"));
+        HadoopHudiColumnValue value = new HadoopHudiColumnValue(ZoneId.of("America/Los_Angeles"));
         value.setField(ColumnType.parseType("ts", "datetimev2(6)"), null);
         value.setRow(new LongWritable(0));
 
         Assert.assertEquals(LocalDateTime.of(1969, 12, 31, 16, 0), value.getDateTime());
     }
 
     @Test
-    public void testInt96TimestampUsesCompatibilityTimezone() {
-        HadoopHudiColumnValue value = new HadoopHudiColumnValue(
-                ZoneId.of("America/Los_Angeles"), ZoneId.of("Asia/Shanghai"));
+    public void testInt96TimestampUsesSessionTimezone() {
+        HadoopHudiColumnValue value = new HadoopHudiColumnValue(ZoneId.of("America/Los_Angeles"));
         value.setField(ColumnType.parseType("ts", "datetimev2(6)"),
                 PrimitiveObjectInspectorFactory.writableTimestampObjectInspector);
         value.setRow(new TimestampWritableV2(Timestamp.ofEpochSecond(0)));
 
-        Assert.assertEquals(LocalDateTime.of(1970, 1, 1, 8, 0), value.getDateTime());
+        Assert.assertEquals(LocalDateTime.of(1969, 12, 31, 16, 0), value.getDateTime());
     }
 }