HappenLee commented on code in PR #23236:
URL: https://github.com/apache/doris/pull/23236#discussion_r1308349211
##########
be/src/vec/sink/vtablet_sink.cpp:
##########
@@ -1328,30 +1538,114 @@ Status VOlapTableSink::send(RuntimeState* state,
vectorized::Block* input_block,
_row_distribution_watch.start();
auto num_rows = block->rows();
size_t partition_num = _vpartition->get_partitions().size();
- if (partition_num == 1 && _tablet_finder->is_find_tablet_every_sink()) {
+ if (!_vpartition->is_auto_partition() && partition_num == 1 &&
+ _tablet_finder->is_find_tablet_every_sink()) {
RETURN_IF_ERROR(_single_partition_generate(state, block.get(),
channel_to_payload, num_rows,
has_filtered_rows));
} else {
- for (int i = 0; i < num_rows; ++i) {
- if (UNLIKELY(has_filtered_rows) &&
_block_convertor->filter_bitmap().Get(i)) {
- continue;
+ if (_vpartition->is_auto_partition()) {
+ std::vector<uint16_t> partition_keys =
_vpartition->get_partition_keys();
+ //TODO: use loop to create missing_vals for multi column.
+ CHECK(partition_keys.size() == 1)
+ << "now support only 1 partition column for auto
partitions.";
+ auto partition_col = block->get_by_position(partition_keys[0]);
+
+ auto missing_vals_col = partition_col.clone_empty();
+
+ // try to find tablet and save missing value
+ for (int i = 0; i < num_rows; ++i) {
+ if (UNLIKELY(has_filtered_rows) &&
_block_convertor->filter_bitmap().Get(i)) {
+ continue;
+ }
+ const VOlapTablePartition* partition = nullptr;
+ bool is_continue = false;
+ uint32_t tablet_index = 0;
+ bool missing_this = false;
+ RETURN_IF_ERROR(_tablet_finder->find_tablet(state,
block.get(), i, &partition,
+ tablet_index,
stop_processing,
+ is_continue,
&missing_this));
+ if (missing_this) {
+
missing_vals_col.column->assume_mutable()->insert_from(*partition_col.column,
+ i);
+ } else {
+ _generate_row_distribution_payload(channel_to_payload,
partition, tablet_index,
+ i, 1);
+ }
}
- const VOlapTablePartition* partition = nullptr;
- bool is_continue = false;
- uint32_t tablet_index = 0;
- RETURN_IF_ERROR(_tablet_finder->find_tablet(
- state, block.get(), i, &partition, tablet_index,
stop_processing, is_continue));
- if (is_continue) {
- continue;
+
+ // for missing partition keys, calc the missing partition and save
in _partitions_need_create
+ auto type = missing_vals_col.type;
+ if (missing_vals_col.column->size() > 0) {
+ vectorized::Block tmp_block = {missing_vals_col};
+ auto [part_ctx, part_func] = _get_partition_function();
+ int result_idx;
+ // calc the start value of missing partition ranges.
+ part_func->prepare(_state, *_output_row_desc, part_ctx.get());
+ part_func->execute(part_ctx.get(), &tmp_block, &result_idx);
+ VLOG_DEBUG << tmp_block.dump_data();
+
+ char convert_buffer[30];
+ // expose the data column
+ vectorized::ColumnPtr col =
tmp_block.get_by_position(result_idx).column;
+ if (auto* nullable =
check_and_get_column<vectorized::ColumnNullable>(*col)) {
+ col = nullable->get_nested_column_ptr();
+ }
+ // calc the end value and save them.
+ if (type->get_type_id() == vectorized::TypeIndex::DateV2) {
+ auto* arg_col = assert_cast<const
vectorized::ColumnDateV2*>(col.get());
+ for (int row = 0; row < arg_col->size(); row++) {
+
_save_a_missing_val<vectorized::DateV2Value<vectorized::DateV2ValueType>>(
+ arg_col, row, convert_buffer,
PrimitiveType::TYPE_DATEV2);
+ }
+ } else if (type->get_type_id() ==
vectorized::TypeIndex::DateTimeV2) {
+ auto* arg_col = assert_cast<const
vectorized::ColumnDateTimeV2*>(col.get());
+ for (int row = 0; row < arg_col->size(); row++) {
+ _save_a_missing_val<
+
vectorized::DateV2Value<vectorized::DateTimeV2ValueType>>(
+ arg_col, row, convert_buffer,
PrimitiveType::TYPE_DATETIMEV2);
+ }
+ } else if (type->get_type_id() == vectorized::TypeIndex::Date)
{
+ auto* arg_col = assert_cast<const
vectorized::ColumnDate*>(col.get());
+ for (int row = 0; row < arg_col->size(); row++) {
+ _save_a_missing_val<vectorized::VecDateTimeValue>(
+ arg_col, row, convert_buffer,
PrimitiveType::TYPE_DATE);
+ }
+ } else if (type->get_type_id() ==
vectorized::TypeIndex::DateTime) {
+ auto* arg_col = assert_cast<const
vectorized::ColumnDateTime*>(col.get());
+ for (int row = 0; row < arg_col->size(); row++) {
+ _save_a_missing_val<vectorized::VecDateTimeValue>(
+ arg_col, row, convert_buffer,
PrimitiveType::TYPE_DATETIME);
+ }
+ }
+ // then call FE to create it. then FragmentExecutor will redo
the load.
+ _automatic_create_partition();
Review Comment:
here need to check the ` _automatic_create_partition` status
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
