Tracking Issue: migrate Arrow/Parquet to the official implementation.

[waynexia]

What type of enhancement is this?

Tech debt reduction

What does the enhancement do?

As discussed in #388, we decide to migrate the arrow/parquet implementation to the official version. This issue tracks the progress of this work.

Implementation challenges

Major API changes

• DataFusion
  • ExecutionPlan::execute change to sync https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlan.html#tymethod.execute

    	  fn execute(
    	      &self,
    	      partition: usize,
    	      context: Arc<TaskContext>
    	  ) -> Result<SendableRecordBatchStream>

  • new interfaces of ExecutionPlan

    • 		  fn output_ordering(&self) -> Option<&[PhysicalSortExpr]>

    • 		  fn statistics(&self) -> Statistics

  • new TaskContext https://docs.rs/datafusion/latest/datafusion/execution/context/struct.TaskContext.html
    • (wrapper over RuntimeEnv, with more functionality)
  • TableProvider is separate into TableProvider(physical stage) and TableSource(logical stage) https://docs.rs/datafusion-expr/latest/datafusion_expr/trait.TableSource.html
• Arrow
  • Array trait
    • apache: https://docs.rs/arrow/latest/arrow/array/trait.Array.html
    • arrow2 (v0.10): https://docs.rs/arrow2/0.10.1/arrow2/array/trait.Array.html
    • the major difference is around validity

      		  fn validity(&self) -> Option<&Bitmap>
      		  
      		  fn with_validity(&self, validity: Option<Bitmap>) -> Box<dyn Array>

  • arrow-format https://github.com/DataEngineeringLabs/arrow-format also needs to migrate to Apache Arrow
• Parquet
  • Mainly used in ParquetWriter (read/write sst) and statistics (predicate).

Works before migration

• Change ExecutionPlan::execute to sync method @waynexia feat: lazy evaluated record batch stream #573
• Specify output_ordering for our ExecutionPlan impl
• Adapt TableSource @v0y4g3r
• (TBD: TaskContext)
• add proxy method like get_validity for ArrayRef
• (TBD: parquet part)
• use datatypes::arrow instead of arrow directly @evenyag refactor: Use re-exported arrow mod from datatypes crate #571
• eliminate the usage of MutableBitmap @evenyag refactor: replace some usage of MutableBitmap by BitVec #610

Migrating

Working on branch replace-arrow2

Target versions:

• DataFusion: 14.0.0
• Arrow: 26.0.0
• Parquet: 26.0.0

Migrating datatypes

Branch datatypes2 is working on migrating our vectors to arrow

• PrimitiveVector @evenyag feat: Implements PrimitiveType and PrimitiveVector for datatypes2 #633
• DateVector @evenyag feat: Implements DateType and DateVector #651
• DateTimeVector @v0y4g3r feat: datatypes2 datetime #661
• TimestampVector @v0y4g3r feat: impl datatypes2/timestamp #686
• StringVector @waynexia feat: Implements StringType and StringVector #659
• ConstantVector @waynexia feat: impl constant vector #680
• NullVector @evenyag feat: Implements NullType and NullVector #658
• ListVector @evenyag feat: Implements ListType and ListVector #681
• Validity @evenyag feat: Implements Validity #684
• ColumnSchema and Schema @evenyag feat: Implements schema for datatypes2 #695

TODO

• Implements VectorOp::cast() feat: impl insert data from query #1025
• Implements VectorOp::take()
• refactor create_current_timestamp_vector()
• Refactor ListValue and ListValueRef
• Better distinguishes logical type and physical type
• Avoid using match to handle timestamp data type
• Refactor common_time::Timestamp using generic #754
• Document that describes the design
• Remove version from Schema Version could be removed from the schema #349
• Refactor RecordBatch and remove arrow_array_get()
• Refactor Timestamp, maybe put all time unit as a flat enum variant
• Refactor ConcreteDataType, move is_xxx to properties
• Test JSON serialization output of values, datatypes, schema, column schema...
• Refactor usage of Helper::static_cast(), avoid checking whether the vector is const in user codes
• More tests for parquet reader
• Consider using ordered float as wrapper type for float vectors.
• Move tests in query/tests to query/src #971
• script/python/builtins/mod.rs move to script/python/builtins.rs, also check other mods refactor: rename some mod.rs to <MOD_NAME>.rs #784
• Refactor script mod to use our RecordBatch/Vector/Schema
• Refactor push_value_ref() method in MutableVector trait #978

Migrating functions

• diff feat: Implements diff accumulator using WrapperType #715
• argmax @evenyag fix: Fix compiler errors in argmax/rate/median/norm_cdf #716
• argmin @evenyag
• mean @waynexia fix: fix compile error for mean/polyval/pow/interp ops #717
• median @evenyag
• percentile @evenyag
• polyval @waynexia fix: fix compile error for mean/polyval/pow/interp ops #717
• scipy_stats_norm_cdf @evenyag
• scipy_stats_norm_pdf @evenyag
• pow @waynexia fix: fix compile error for mean/polyval/pow/interp ops #717
• rate @evenyag
• clip @evenyag
• interp @waynexia fix: fix compile error for mean/polyval/pow/interp ops #717

TODO

• test test_clip_fn_* may fail with SIGSEVG fix: pre-cast to avoid tremendous match arms #734

[evenyag]

What about Field and Schema, I remember that arrow2 and arrow have different APIs.

[evenyag]

add proxy method like get_validity for ArrayRef

Only a few places use get_validity(), I think we could skip this step.

BTW, we also use methods in arrow::compute

greptimedb/src/script/src/python/vector.rs

Lines 115 to 122 in 6d762aa

	Ok(arrow::compute::cast::cast(
	arithmetics::div(tmp.as_ref(), arr).as_ref(),
	&DataType::Int64,
	cast::CastOptions {
	wrapped: false,
	partial: true,
	},
	)

They might have different semantic

• arrow2 CastOptions
• arrow CastOptions

[waynexia]

They might have different semantic

* [arrow2 CastOptions](https://docs.rs/arrow2/latest/arrow2/compute/cast/struct.CastOptions.html)

* [arrow CastOptions](https://docs.rs/arrow/latest/arrow/compute/struct.CastOptions.html)

Yes, so many API changes. For this particular case, I think we can set safe to true in the new option.

Gladly (or maybe sadly 🥲) it looks like we don't have this kind of boundary-case tests. I almost see the future that we've paid lots of care to review the change (I hope so 🤪) but still ignore some critical changes that will take us a day to trace it down.

[v0y4g3r]

We may also need to refactor Vector to separate vector's logical type from it's internal representation like discussed in #203

[waynexia]

We may also need to refactor Vector to separate vector's logical type from it's internal representation like discussed in #203

Looks #203 is suggesting a non-trivial change. We can dive into it after this migration.

[evenyag]

I'm working on the new vectors based on the official arrow in branch datatypes2. Now I implement the boolean and binary vector. The next step is porting the primitive vector. @waynexia @v0y4g3r

[evenyag]

• test test_clip_fn_* may fail with SIGSEVG

I reproduced this BUG in another repo and find out that it is highly related to the usage of the with_match_primitive_type_id macro.

The AddressSanitizer also says that the stack overflowed when running the tests.

=================================================================
==71205==ERROR: AddressSanitizer: stack-buffer-overflow on address 0x00016d433f60 at pc 0x000102ca8210 bp 0x00016d3caf90 sp 0x00016d3caf88
WRITE of size 8 at 0x00016d433f60 thread T2
    #0 0x102ca820c in debug_datatypes::macros::tests::eval_i64::h9032765d623f5419 macros.rs:178

The define_eval expands to about 11 * 11 * 11 = 1331 (there are 10 primitive types and an unreachable branch) match arms. The overflow should be caused by rust-lang/rust#34283. Thanks for providing this helpful hint @discord9.

[evenyag]

arrow only provides a generic version of arithmetic operation, but arrow2 supports passing scalar dynamically

arrow:

pub fn add_scalar_dyn<T>(array: &dyn Array, scalar: T::Native) -> Result<ArrayRef>
where
    T: ArrowNumericType,
    T::Native: ArrowNativeTypeOp,
{}

arrow2:

pub fn add_scalar(lhs: &dyn Array, rhs: &dyn Scalar) -> Box<dyn Array> {}

[DiamondMofeng]

Is this error relavate to this issue?
This function is not necessary, but it is useful in time serise query.

[DiamondMofeng]

Sorry, I didn't see there is an available now() function, which is an alternative.

[evenyag]

This function is not necessary, but it is useful in time serise query

Unfortunately, we don't support Time64 data type now. Note that the Timestamp type is different from the Time type.

[killme2008]

What's the status of this tracking issue right now? Can we close it or keep it? @evenyag @v0y4g3r

[v0y4g3r]

What's the status of this tracking issue right now? Can we close it or keep it? @evenyag @v0y4g3r

I think it's sufficient to close this tracking issue. The unresolved todos mainly relate to data type refactoring, it's better to open another issue for that topic.