refactor(source): enhance parquet file source #19221

wcy-fdu · 2024-10-31T13:51:02Z

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

This PR primarily focuses on the conversion of the Timestamp type. The Arrow Timestamp includes four types:


pub enum TimeUnit {
    /// Time in seconds.
    Second,
    /// Time in milliseconds.
    Millisecond,
    /// Time in microseconds.
    Microsecond,
    /// Time in nanoseconds.
    Nanosecond,
}

but previously we only handled one Microsecond. This pr makes the comparison of data types more reasonable:

when doing data type check at column pruning, make risingwave_common::types::DataType::Timestamp match arrow_schema_iceberg::DataType::Timestamp(_, _).
when convert arrow field to rw field, add Nanosecond, Millisecond, Second type.

Checklist

I have written necessary rustdoc comments
I have added necessary unit tests and integration tests
I have added test labels as necessary. See details.
I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
All checks passed in ./risedev check (or alias, ./risedev c)
My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

src/connector/src/source/filesystem/opendal_source/opendal_reader.rs

chenzl25 · 2024-11-01T08:12:23Z

src/common/src/array/arrow/arrow_impl.rs

+    fn from_timestampsecond_array(
+        &self,
+        array: &arrow_array::TimestampSecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+    fn from_timestampsecond_some_array(
+        &self,
+        array: &arrow_array::TimestampSecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+
+    fn from_timestampms_array(
+        &self,
+        array: &arrow_array::TimestampMillisecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+
+    fn from_timestampms_some_array(
+        &self,
+        array: &arrow_array::TimestampMillisecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamptz(array.into()))
+    }


I am curious, we don't need to do any conversion between these types? All of them can be implemented in the same way?

No, just added such logic and verify it works. I constructed a parquet file containing timestamps of four time units, then imported it into RisingWave and it was parsed successfully:

I also don't get why we need to add different methods to convert timestamp in different units. IIUC, the current conversion logic won't lose any precision. What happened if we just add different units of the timestamp type to converts macro and rely on the original implementation to do the conversion?

converts!(TimestampArray, arrow_array::TimestampMillisecondArray, @map); converts!(TimestampArray, arrow_array::TimestampSecondArray, @map); converts!(TimestampArray, arrow_array::TimestampNanosecondArray, @map);

In our previous code, we only handled timestamps in microseconds, so this conversion is only correct for timestamp(MicroSecond, _). As you mentioned in the logic, the seconds part is % 1_000_000, and the nanoseconds part is also % 1_000_000. If all four time units are converted using this logic, it will yield incorrect results, as follows:

The correct approach should be as shown in the PR, where different handling is done based on the time unit. For example, for milliseconds, it should be %1000 instead of %100000:

TimeUnit::Millisecond => Timestamp( DateTime::from_timestamp((value / 1_000) as _, (value % 1_000 * 1000000) as _) .unwrap() .naive_utc(), ),

So I think specifying the time unit is essential.

src/connector/src/source/filesystem/opendal_source/opendal_reader.rs

src/common/src/array/arrow/arrow_impl.rs

xxchan · 2024-11-06T13:25:37Z

src/common/src/array/arrow/arrow_impl.rs

 /// Converts RisingWave value from and into Arrow value.
 trait FromIntoArrow {
    /// The corresponding element type in the Arrow array.
    type ArrowType;
+    /// The timestamp type used to distinguish different time units, only utilized when the Arrow type is a timestamp.
+    type TimestampType;
    fn from_arrow(value: Self::ArrowType) -> Self;
    fn into_arrow(self) -> Self::ArrowType;
+    /// Used for converting timestamp types and will not be used in conversions of other types.
+    fn from_arrow_with_unit(value: Self::ArrowType, time_unit: Self::TimestampType) -> Self;
 }


Generally speaking, for type conversion, schema (i.e., type information) should be provided, since data itself may not be self-contained.

fn convert(v_from, t_from, t_to) -> v_to

t_from & t_to are not always needed though

e.g., our Avro conversion

risingwave/src/connector/codec/src/decoder/avro/mod.rs

Lines 89 to 93 in 9a32e75

pub fn convert_to_datum<'b>(

&self,

value: &'b Value,

type_expected: &DataType,

) -> AccessResult<DatumCow<'b>>

Therefore, _with_unit looks a little strange to me. Haven't checked the details of Arrow yet.

cc @xiangjinwu @BugenZhao might also be able to comment

https://docs.rs/arrow-schema/53.2.0/arrow_schema/enum.DataType.html
If we need _with_unit to pass extra info from DataType, how about the following?

IntervalUnit

i32 length for FixedSizeBinary

precision and scale for Decimal{128,256}

To repeat, DataType should be provided as data itself may not be self-contained.

e2e_test/s3/fs_sink.py

stdrc · 2024-11-08T06:27:53Z

src/common/src/array/arrow/arrow_impl.rs

+    ($ArrayType:ty, $ArrowType:ty, $time_unit:expr, @map) => {
+
+        impl From<&$ArrayType> for $ArrowType {
+            fn from(array: &$ArrayType) -> Self {
+                array.iter().map(|o| o.map(|v| v.into_arrow_with_unit($time_unit))).collect()
+            }
+        }
+
+        impl From<&$ArrowType> for $ArrayType {
+            fn from(array: &$ArrowType) -> Self {
+                array.iter().map(|o| {
+                    o.map(|v| {
+                        let timestamp = <<$ArrayType as Array>::RefItem<'_> as FromIntoArrow>::from_arrow_with_unit(v, $time_unit);
+                        timestamp
+                    })
+                }).collect()
+            }
+        }
+
+        impl From<&[$ArrowType]> for $ArrayType {
+            fn from(arrays: &[$ArrowType]) -> Self {
+                arrays
+                    .iter()
+                    .flat_map(|a| a.iter())
+                    .map(|o| {
+                        o.map(|v| {
+                            <<$ArrayType as Array>::RefItem<'_> as FromIntoArrow>::from_arrow(v)
+                        })
+                    })
+                    .collect()
+            }
+        }
+
+    };


This looks very specialized. What about putting it in a new macro, like converts_timestamp?

stdrc · 2024-11-08T06:29:14Z

src/common/src/array/arrow/arrow_impl.rs

+    /// The timestamp type used to distinguish different time units, only utilized when the Arrow type is a timestamp.
+    type TimestampType;


Why is this needed? I guess it's too intrusive to put timestamp related things in every type?

We need to distinguish the four timestamp types of arrow, instead of just using one i64 to represent them.

hzxa21 · 2024-11-11T03:10:57Z

src/common/src/array/arrow/arrow_impl.rs

            Timestamp(Microsecond, None) => DataType::Timestamp,
            Timestamp(Microsecond, Some(_)) => DataType::Timestamptz,
+            Timestamp(Second, None) => DataType::Timestamp,
+            Timestamp(Second, Some(_)) => DataType::Timestamptz,
+            Timestamp(Millisecond, None) => DataType::Timestamp,
+            Timestamp(Millisecond, Some(_)) => DataType::Timestamptz,
+            Timestamp(Nanosecond, None) => DataType::Timestamp,
+            Timestamp(Nanosecond, Some(_)) => DataType::Timestamptz,


Timestamp(_, None) => DataType::Timestamp, Timestamp(_, Some(_)) => DataType::Timestamptz,

prefer to keep time unit designated as a reminder, there are four types of timeunit that need to be handled. Previously, other types were ignored because of the use of _.

Previously we use concrete type Microsecond instead of _ so it was ignored not because of the use of _ but the contrast. I don't have a strong opinion on this so listing the types explicitly here is also okay. I am more concerned about having separate logics to handle the different time unit separately given that our original implementation won't lose any precision. See my comment below.

won't lose any precision

explained in #19221 (comment)

hzxa21 · 2024-11-11T03:46:56Z

src/common/src/array/arrow/arrow_impl.rs

+    fn from_timestampsecond_array(
+        &self,
+        array: &arrow_array::TimestampSecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+    fn from_timestampsecond_some_array(
+        &self,
+        array: &arrow_array::TimestampSecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+
+    fn from_timestampms_array(
+        &self,
+        array: &arrow_array::TimestampMillisecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamp(array.into()))
+    }
+
+    fn from_timestampms_some_array(
+        &self,
+        array: &arrow_array::TimestampMillisecondArray,
+    ) -> Result<ArrayImpl, ArrayError> {
+        Ok(ArrayImpl::Timestamptz(array.into()))
+    }


I also don't get why we need to add different methods to convert timestamp in different units. IIUC, the current conversion logic won't lose any precision. What happened if we just add different units of the timestamp type to converts macro and rely on the original implementation to do the conversion?

converts!(TimestampArray, arrow_array::TimestampMillisecondArray, @map); converts!(TimestampArray, arrow_array::TimestampSecondArray, @map); converts!(TimestampArray, arrow_array::TimestampNanosecondArray, @map);

wcy-fdu · 2024-11-11T09:30:42Z

In order not to invade the FromToArrow trait, the timestamp conversion is split into another trait FromIntoArrowWithUnit, and a new macro converts_with_timeunit! is also added, which is much clearer than before. As for why it is necessary to formulate a time unit, see #19221 (comment)

src/common/src/array/arrow/arrow_impl.rs

hzxa21 · 2024-11-11T10:30:45Z

src/common/src/array/arrow/arrow_impl.rs

+            TimeUnit::Second => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_seconds(),
+            TimeUnit::Millisecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_milliseconds(),
+            TimeUnit::Microsecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_microseconds()
+                .unwrap(),
+            TimeUnit::Nanosecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_nanoseconds()
+                .unwrap(),


I think you can use DateTime::timestamp_xxx() here: https://doc.servo.org/chrono/struct.DateTime.html#method.timestamp_micros

This is not resolved.

…nto wcy/fix_parquet_type

hzxa21

Rest LGTM

hzxa21 · 2024-11-12T02:12:49Z

src/common/src/array/arrow/arrow_impl.rs

+            TimeUnit::Second => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_seconds(),
+            TimeUnit::Millisecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_milliseconds(),
+            TimeUnit::Microsecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_microseconds()
+                .unwrap(),
+            TimeUnit::Nanosecond => self
+                .0
+                .signed_duration_since(NaiveDateTime::default())
+                .num_nanoseconds()
+                .unwrap(),


This is not resolved.

hzxa21 · 2024-11-12T02:18:29Z

src/common/src/array/arrow/arrow_impl.rs

+            }
+            Timestamp(Millisecond, Some(_)) => {
+                self.from_timestampms_some_array(array.as_any().downcast_ref().unwrap())
+            }
            Timestamp(Microsecond, None) => {
                self.from_timestampus_array(array.as_any().downcast_ref().unwrap())
            }
            Timestamp(Microsecond, Some(_)) => {


Not directly related to this PR but I wonder why we ignore the timezone (Some(_)) from arrow timestamp when constructing Timestamptz

Introduced in #17201

Actually, we have encountered three issues regarding the conversion of the timestamp type:

When converting Arrow's timestamp type to our timestamp, we should decide whether to convert it to timestamp or timestampz based on whether there is a timezone (none or some).

When converting Arrow's timestamp type to our timestamp, only microseconds were converted, while the other three time units were not converted.

In type comparisons, Arrow's timestamp(_, none) should match with rw’s timestamp, while timestamp(_, some) should match with rw’s timestamptz. However, previously, apart from microseconds, the other units did not match.

#17201 fix 1, and this pr fix 2 and 3.

I understand. Correct me if I am wrong, I think 1 is not fixed completely because we don't use the correct timezone for timestamptz based on the arrow data type.

Discussed offline, indeed this is another bug: during the conversion, the timezone should be handled using Arrow's timezone(), as we are currently using the default timezone. I will fix this in the next PR. Thanks to @hzxa21 for identifying this issue.

Double checked, their is no correctness issue:
When the timezone in Arrow is Some(_), the i64 always stores UTC time, which is consistent with PG/RW's timestamptz. Therefore, we only need to distinguish between None and Some(_), as the contents of Some do not affect the actual value; they are merely for display purposes.
refer to https://docs.rs/arrow-schema/53.2.0/arrow_schema/enum.DataType.html#timestamps-with-a-non-empty-timezone

hzxa21 · 2024-11-12T02:19:16Z

src/common/src/array/arrow/arrow_impl.rs

+            TimeUnit::Second => self.timestamp(),
+            TimeUnit::Millisecond => self.timestamp_millis(),
+            TimeUnit::Microsecond => self.timestamp_micros(),
+            TimeUnit::Nanosecond => self.timestamp_nanos().unwrap_or_default(),


Should we use unwrap instead of unwrap_or_default?

I think it's better not to panic when I encounter a problem with the type conversion? It's better to convert to 0 by mistake than to panic a cluster

If our TimestampTz is in microsecond unit, I think we should never overflow?

yes, microsecond checked_mul(1_000) is safe, then unwrap() safely.

… branch release-2.1 (#19401)

enhance parquet source and sink

2301fae

github-actions bot added the type/refactor label Oct 31, 2024

wcy-fdu commented Oct 31, 2024

View reviewed changes

src/connector/src/source/filesystem/opendal_source/opendal_reader.rs Outdated Show resolved Hide resolved

graphite-app bot requested review from a team October 31, 2024 14:23

minor

c86a8ce

wcy-fdu requested review from hzxa21 and chenzl25 November 1, 2024 06:20

chenzl25 requested a review from xxchan November 1, 2024 08:07

chenzl25 reviewed Nov 1, 2024

View reviewed changes

xxchan reviewed Nov 1, 2024

View reviewed changes

src/connector/src/source/filesystem/opendal_source/opendal_reader.rs Outdated Show resolved Hide resolved

wcy-fdu added 5 commits November 3, 2024 16:48

handle Map type

a1d1a9b

fix convert fro nanos, millis, secound

d804208

fix Millisecond

01d102c

fmt

7c92a3f

add some comments

61e79a1

chenzl25 reviewed Nov 6, 2024

View reviewed changes

src/common/src/array/arrow/arrow_impl.rs Outdated Show resolved Hide resolved

xxchan reviewed Nov 6, 2024

View reviewed changes

xxchan requested review from xiangjinwu and BugenZhao November 6, 2024 13:30

wcy-fdu added 2 commits November 7, 2024 16:10

add into_arrow_with_unit

69c8320

minor

857d86d

stdrc reviewed Nov 8, 2024

View reviewed changes

hzxa21 reviewed Nov 11, 2024

View reviewed changes

split into two macros and trait

8d033f7

wcy-fdu requested review from hzxa21 and chenzl25 November 11, 2024 09:30

hzxa21 reviewed Nov 11, 2024

View reviewed changes

minor

c83f421

Merge branch 'main' of https://github.com/risingwavelabs/risingwave i…

99b6846

…nto wcy/fix_parquet_type

wcy-fdu requested a review from hzxa21 November 12, 2024 02:17

hzxa21 approved these changes Nov 12, 2024

View reviewed changes

minor

4006d13

wcy-fdu enabled auto-merge November 12, 2024 03:19

wcy-fdu disabled auto-merge November 12, 2024 03:22

wcy-fdu added this pull request to the merge queue Nov 12, 2024

Merged via the queue into main with commit b141881 Nov 12, 2024
28 of 29 checks passed

wcy-fdu deleted the wcy/fix_parquet_type branch November 12, 2024 04:23

wcy-fdu added the need-cherry-pick-release-2.1 label Nov 13, 2024

github-actions bot mentioned this pull request Nov 13, 2024

cherry-pick refactor(source): enhance parquet file source (#19221) to branch release-2.1 #19368

Open

wcy-fdu added a commit that referenced this pull request Nov 15, 2024

refactor(source): enhance parquet file source (#19221)

be6c273

hzxa21 mentioned this pull request Nov 15, 2024

cherry-pick refactor(source): enhance parquet file source (#19221) to branch release-2.1 #19401

Merged

9 tasks

github-merge-queue bot pushed a commit that referenced this pull request Nov 15, 2024

cherry-pick refactor(source): enhance parquet file source (#19221) to…

9988269

… branch release-2.1 (#19401)

wcy-fdu mentioned this pull request Nov 28, 2024

cherry-pick feat(connector): support more parquet data types (#19561) to branch release-2.0 #19602

Closed

9 tasks

lmatz mentioned this pull request Dec 3, 2024

feat: support ingesting Timestamp(nanoseconds) from Iceberg #19163

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(source): enhance parquet file source #19221

refactor(source): enhance parquet file source #19221

wcy-fdu commented Oct 31, 2024

chenzl25 Nov 1, 2024

wcy-fdu Nov 5, 2024

hzxa21 Nov 11, 2024

wcy-fdu Nov 11, 2024

wcy-fdu Nov 11, 2024

xxchan Nov 6, 2024

xxchan Nov 6, 2024

xxchan Nov 6, 2024

xiangjinwu Nov 7, 2024

stdrc Nov 8, 2024 •

edited

Loading

stdrc Nov 8, 2024

wcy-fdu Nov 8, 2024

hzxa21 Nov 11, 2024

wcy-fdu Nov 11, 2024

hzxa21 Nov 11, 2024 •

edited

Loading

wcy-fdu Nov 11, 2024 •

edited

Loading

hzxa21 Nov 11, 2024

wcy-fdu commented Nov 11, 2024

hzxa21 Nov 11, 2024

hzxa21 Nov 12, 2024

hzxa21 left a comment

hzxa21 Nov 12, 2024

hzxa21 Nov 12, 2024

hzxa21 Nov 12, 2024

wcy-fdu Nov 12, 2024

hzxa21 Nov 12, 2024

wcy-fdu Nov 12, 2024 •

edited

Loading

wcy-fdu Nov 12, 2024

hzxa21 Nov 12, 2024

wcy-fdu Nov 12, 2024

hzxa21 Nov 12, 2024

wcy-fdu Nov 12, 2024

	pub fn convert_to_datum<'b>(
	&self,
	value: &'b Value,
	type_expected: &DataType,
	) -> AccessResult<DatumCow<'b>>

		/// The timestamp type used to distinguish different time units, only utilized when the Arrow type is a timestamp.
		type TimestampType;

refactor(source): enhance parquet file source #19221

refactor(source): enhance parquet file source #19221

Conversation

wcy-fdu commented Oct 31, 2024

What's changed and what's your intention?

Checklist

Documentation

Release note

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stdrc Nov 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzxa21 Nov 11, 2024 • edited Loading

Choose a reason for hiding this comment

wcy-fdu Nov 11, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcy-fdu commented Nov 11, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hzxa21 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wcy-fdu Nov 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stdrc Nov 8, 2024 •

edited

Loading

hzxa21 Nov 11, 2024 •

edited

Loading

wcy-fdu Nov 11, 2024 •

edited

Loading

wcy-fdu Nov 12, 2024 •

edited

Loading