[GLUTEN-8492][CH] Offload RangeExec

[taiyang-li]

What changes were proposed in this pull request?

(Please fill in changes proposed in this fix)

(Fixes: #8492)

How was this patch tested?

All existing uts using range function. Below ut is excluded because assertion that plan contains RangeExec operator would fail after RangeExec is offloaded to gluten.

• "SPARK-27439: Explain result should match collected result after view change"

[github-actions]

Thanks for opening a pull request!

Could you open an issue for this pull request on Github Issues?

https://github.com/apache/incubator-gluten/issues

Then could you also rename commit message and pull request title in the following format?

[GLUTEN-${ISSUES_ID}][COMPONENT]feat/fix: ${detailed message}

See also:

• Other pull requests

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

#8492

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[github-actions]

Run Gluten Clickhouse CI on x86

[taiyang-li]

For performance comparison refer to #8492 (comment)

[github-actions]

Run Gluten Clickhouse CI on x86

[taiyang-li]

@CodiumAI-Agent /review

[CodiumAI-Agent]

PR Reviewer Guide 🔍

(Review updated until commit febd663)

Here are some key observations to aid the review process:

🎫 Ticket compliance analysis 🔶 8492 - Partially compliant Compliant requirements: Investigate and address the performance issue with the sum function using a filter in the ClickHouse backend compared to vanilla Spark. Implement necessary changes to offload RangeExec to the Gluten backend. Ensure compatibility and correctness of the solution with existing functionalities. Non-compliant requirements: Update relevant tests to reflect the changes. Requires further human verification: Validate the performance improvement of the sum function with a filter in real-world scenarios. Ensure the correctness of the offloaded RangeExec through extensive testing.

⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪

🧪 No relevant tests

🔒 No security concerns identified

⚡ Recommended focus areas for review Possible Issue The doTransform method in CHRangeExecTransformer introduces a new transformation logic. Ensure that the generated Substrait plan is correct and compatible with the ClickHouse backend. override def doTransform(context: SubstraitContext): TransformContext = { val output = outputAttributes val typeNodes = ConverterUtils.collectAttributeTypeNodes(output) val nameList = ConverterUtils.collectAttributeNamesWithoutExprId(output) val columnTypeNodes = JavaConverters .seqAsJavaListConverter( output.map(attr => new ColumnTypeNode(NamedStruct.ColumnType.NORMAL_COL))) .asJava val optimizationContent = s"isRange=1\n" val optimization = BackendsApiManager.getTransformerApiInstance.packPBMessage( StringValue.newBuilder.setValue(optimizationContent).build) val extensionNode = ExtensionBuilder.makeAdvancedExtension(optimization, null) val readNode = RelBuilder.makeReadRel( typeNodes, nameList, columnTypeNodes, null, extensionNode, context, context.nextOperatorId(this.nodeName)) TransformContext(output, readNode) Performance Concern The SourceFromRange implementation may have performance implications due to its handling of large ranges and potential overflows. Verify its efficiency and correctness. SourceFromRange::SourceFromRange(const DB::Block & header_, Int64 start_, Int64 end_, Int64 step_, Int32 num_slices_, Int32 slice_index_, size_t max_block_size_) : DB::ISource(header_) , start(start_) , end(end_) , step(step_) , num_slices(num_slices_) , slice_index(slice_index_) , max_block_size(max_block_size_) , num_elements(getNumElements()) , is_empty_range(start == end ) { const auto & header = getOutputs().front().getHeader(); if (header.columns() != 1) throw Exception(ErrorCodes::NUMBER_OF_COLUMNS_DOESNT_MATCH, "Expected 1 column, got {}", header.columns()); if (!header.getByPosition(0).type->equals(DataTypeInt64())) throw Exception(ErrorCodes::TYPE_MISMATCH, "Expected Int64 column, got {}", header.getByPosition(0).type->getName()); if (step == 0) throw Exception(ErrorCodes::LOGICAL_ERROR, "Step cannot be zero"); Int128 partition_start = (slice_index * num_elements) / num_slices * step + start; Int128 partition_end = (((slice_index + 1) * num_elements) / num_slices) * step + start; auto get_safe_margin = [](Int128 bi) -> Int64 { if (bi <= std::numeric_limits<Int64>::max() && bi >= std::numeric_limits<Int64>::min()) return static_cast<Int64>(bi); else if (bi > 0) return std::numeric_limits<Int64>::max(); else return std::numeric_limits<Int64>::min(); }; safe_partition_start = get_safe_margin(partition_start); safe_partition_end = get_safe_margin(partition_end); current = safe_partition_start; previous = 0; overflow = false; } Int128 SourceFromRange::getNumElements() const { const auto safe_start = static_cast<Int128>(start); const auto safe_end = static_cast<Int128>(end); if ((safe_end - safe_start) % step == 0 || (safe_end > safe_start) != (step > 0)) { return (safe_end - safe_start) / step; } else { // the remainder has the same sign with range, could add 1 more return (safe_end - safe_start) / step + 1; } } DB::Chunk SourceFromRange::generate() { if (is_empty_range) return {}; if (overflow || (step > 0 && current >= safe_partition_end) || (step < 0 && current <= safe_partition_end)) return {}; auto column = DB::ColumnInt64::create(); auto & data = column->getData(); data.resize_exact(max_block_size); size_t row_i = 0; if (step > 0) { for (; current < safe_partition_end && !overflow && row_i < max_block_size; ++row_i) { previous = current; data[row_i] = current; current += step; overflow = current < previous; } } else { for (; current > safe_partition_end && !overflow && row_i < max_block_size; ++row_i) { previous = current; data[row_i] = current; current += step; overflow = current > previous; } } data.resize_exact(row_i); // std::cout << "gen rows:" << column->size() << std::endl; DB::Columns columns; columns.push_back(std::move(column)); return DB::Chunk(std::move(columns), row_i); } Serialization Logic The new makeExtensionTable method for range handling introduces JSON serialization. Ensure that the serialization and deserialization processes are robust and error-free. public static ExtensionTableNode makeExtensionTable( Long start, Long end, Long step, Integer numSlices, Integer sliceIndex) throws JsonProcessingException { ObjectMapper mapper = new ObjectMapper(); ObjectNode json = mapper.createObjectNode(); json.put("start", start); json.put("end", end); json.put("step", step); json.put("numSlices", numSlices); json.put("sliceIndex", sliceIndex); String result; result = mapper.writeValueAsString(json); return new ExtensionTableNode( Collections.emptyList(), result, (Seq<String>) Seq$.MODULE$.<String>empty()); }

[github-actions]

Run Gluten Clickhouse CI on x86

[CodiumAI-Agent]

Persistent review updated to latest commit febd663

[taiyang-li]

Run Gluten Clickhouse CI on x86

[PHILO-HE]

Mainly reviewed the common code. Basically looks good!

[PHILO-HE]

Maybe, we can move the existing GlutenSQLRangeExecSuite.scala from spark33 ut folder to gluten-ut/test folder, which can get it tested for all Spark versions and both backends.

[github-actions]

Run Gluten Clickhouse CI on x86

[taiyang-li]

Maybe, we can move the existing GlutenSQLRangeExecSuite.scala from spark33 ut folder to gluten-ut/test folder, which can get it tested for all Spark versions and both backends.

done.